import uuid
from fastapi.encoders import jsonable_encoder
from fastapi.responses import StreamingResponse
from fastapi.concurrency import run_in_threadpool
from pathlib import Path as FilePath
from sqlalchemy import create_engine, text, and_, func, or_, delete, insert
from app.datatk.data_clear import EnhancedDataCleaner
from app.schemas.schemas import UserMsgRead, UserSessionRead
from app.datatk.data_help import DataHelp
from app.ai.ai_getdata import AIGetData
import pandas as pd
from app.models.user import auth_backend, current_active_user, fastapi_users
from sqlalchemy.orm import sessionmaker
from typing import List, Optional, Dict, Any
from sqlalchemy.future import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from app.models.db import DataSet, Task, UserMsg, User, create_df_table, drop_table, get_async_session, create_db_and_tables, UserSession,sync_engine, engine, AnalysisDimension, RoleDatasetAccess, UserRole, Role, DataSource, DataField, SmsCode, AnalysisReport
from app.ai.ai_prompt import AiPrompt
from openai import OpenAI
import numpy as np
import random
from fastapi import APIRouter, BackgroundTasks, Depends, FastAPI, File, Form, HTTPException, UploadFile, Query, Path, Request, Body
from loguru import logger
from datetime import datetime, time, timedelta
import io
import json
import os
import asyncio
from app.services.dataset_service import DatasetService
from config import AI_API_KEY, DATABASETB_URL, LOG_PATH, DATABASE_URL, AI_BASEURL
# 创建异步会话，需要修正为使用异步会话工厂
from app.models.db import async_session_maker
from sqlalchemy.exc import SQLAlchemyError
from app.utils.error_handler import FileUploadException, DataAnalysisException, DatabaseException
from app.utils.auth import get_user_role_codes
from app.services.task_executor import TaskExecutor
from app.ai.data_processor import DataProcessor
# 导入token检查工具
from app.utils.token_utils import check_prompt_limit, get_prompt_info

# 智能助手相关导入
from app.ai.intelligent_assistant_engine import IntelligentAssistantEngine



logger.add(
    sink=LOG_PATH,
    enqueue=True,
    rotation="4 weeks",
    retention="4 months",
    encoding="utf-8",
    backtrace=True,
    diagnose=True,
    compression="zip",
)

router = APIRouter(prefix="/sessions", tags=["sessions"])
# 获取建议
advisor = AIGetData(AI_BASEURL, AI_API_KEY)



# 初始化智能助手引擎
assistant_engine = IntelligentAssistantEngine()


@router.post("/chat_with_assistant", response_model=None)
async def chat_with_assistant(
    message: str = Form(..., description="用户消息内容"),
    session_id: str = Form(None, description="会话ID，可选，为空时创建新会话"),
    session_title: str = Form("智能助手对话", description="会话标题，可选"),
    current_user: User = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_session)
):
    """与智能助手对话，优化版本 - 立即流式响应"""
    try:
        # 如果没有提供session_id，创建新会话
        if not session_id:
            session_id = await assistant_engine.create_new_session(
                str(current_user.id), session_title, db
            )
        
        # 🚀 立即创建用户消息记录
        user_timestamp = datetime.now()
        user_msg = UserMsg(
            user_id=str(current_user.id),
            session_id=session_id,
            message=message,
            message_type='user',
            timestamp=user_timestamp
        )
        db.add(user_msg)
        await db.commit()
        await db.refresh(user_msg)
        user_message_id = str(user_msg.id)
        
        # 🚀 创建系统消息记录（初始为空），时间戳明确晚于用户消息
        from datetime import timedelta
        system_timestamp = user_timestamp + timedelta(seconds=1)
        system_msg = UserMsg(
            user_id=str(current_user.id),
            session_id=session_id,
            message='',  # 初始为空，流式输出完成后更新
            message_type='system',
            message_sourceid=user_message_id,
            timestamp=system_timestamp
        )
        db.add(system_msg)
        await db.commit()
        await db.refresh(system_msg)

        async def optimized_stream():
            # 🎯 关键修改：分离进度信息和实际回复内容
            actual_reply_content = ""  # 只存储实际的AI回复内容，用于保存到数据库
            
            try:
                
                # 🚀 第三步：快速构建基础上下文（避免复杂查询）
                try:
                    # 只查询最近3条消息，减少数据库查询时间
                    recent_messages_query = select(UserMsg).where(
                        UserMsg.session_id == session_id,
                        UserMsg.message_type == 'user'
                    ).order_by(UserMsg.timestamp.desc()).limit(3)
                    
                    recent_messages_result = await db.execute(recent_messages_query)
                    recent_messages = recent_messages_result.scalars().all()
                    
                    # 快速构建上下文
                    conversation_context = {
                        'user_history': [],
                        'recent_topics': [msg.message[:50] for msg in recent_messages if msg.message],  # 只取前50字符
                        'user_preferences': {
                            'detail_level': '标准',
                            'response_style': '友好专业',
                            'chart_preference': []
                        }
                    }
                    
                except Exception as context_error:
                    logger.warning(f"构建上下文失败，使用默认上下文: {str(context_error)}")
                    conversation_context = {
                        'user_history': [],
                        'recent_topics': [message],
                        'user_preferences': {
                            'detail_level': '标准',
                            'response_style': '友好专业',
                            'chart_preference': []
                        }
                    }
                
                # 🚀 第四步：异步处理复杂任务，不阻塞流式输出
                async def process_in_background():
                    try:
                        # 执行智能助手的处理逻辑，使用已创建的用户消息ID
                        result = await assistant_engine.process_user_input_with_existing_message(
                            message, session_id, str(current_user.id), user_message_id, db
                        )
                        
                        return result
                    except Exception as e:
                        logger.error(f"后台处理失败: {str(e)}")
                        return {"error": str(e), "status": "error"}
                
                # 启动后台处理任务
                import asyncio
                background_task = asyncio.create_task(process_in_background())
                
                # 第五步：等待处理完成，但提供进度反馈（进度信息不保存）
                wait_time = 0
                max_wait_time = 15  # 最多等待15秒，减少等待时间
                check_interval = 0.5  # 每0.5秒检查一次
                

                
                while not background_task.done() and wait_time < max_wait_time:
                    await asyncio.sleep(check_interval)
                    wait_time += check_interval
                    

                
                # 🚀 第六步：处理任务结果
                result = None
                if background_task.done():
                    try:
                        result = await background_task
                        
                        if "error" in result:
                            error_msg = f"❌ 处理失败: {result['error']}\n\n"
                            actual_reply_content += error_msg  # 错误信息需要保存
                            yield error_msg
                            
                            # 提供基本的错误回复
                            basic_reply = "抱歉，处理您的请求时遇到了问题。请尝试：\n\n1. 简化您的问题\n2. 检查数据集是否正确选择\n3. 稍后重试\n\n如果问题持续，请联系技术支持。"
                            actual_reply_content += basic_reply  # 基本回复需要保存
                            yield basic_reply
                            return
                        

                        
                    except Exception as e:
                        error_msg = f"❌ 处理异常: {str(e)}\n\n"
                        actual_reply_content += error_msg  # 错误信息需要保存
                        yield error_msg
                        
                        # 提供基本回复
                        basic_reply = "处理出现异常，请重新提问或简化问题。"
                        actual_reply_content += basic_reply  # 基本回复需要保存
                        yield basic_reply
                        return
                else:
                    # 超时处理 - 提供更好的用户体验
                    
                    # 尝试取消后台任务并提供基本回复
                    if not background_task.done():
                        background_task.cancel()
                    
                    basic_reply = "为了提供更好的体验，建议您：\n\n1. 🎯 **简化问题**：尝试询问更具体的问题\n2. 📊 **分步查询**：将复杂问题拆分为几个简单问题\n3. 💡 **举例说明**：提供具体的查询示例\n\n我们正在后台优化处理速度，感谢您的耐心！"
                    actual_reply_content += basic_reply  # 基本回复需要保存
                    yield basic_reply
                    return
                
                # 🚀 第七步：流式生成AI回复（这部分是实际回复内容，需要保存）
                if result:
                    try:
                        def generate_sync():
                            yield from advisor.analyze_data_with_ai(
                                message,
                                result,
                                conversation_context
                            )

                        # 流式输出AI生成的内容，并保存到actual_reply_content
                        for chunk in generate_sync():
                            actual_reply_content += chunk  # 🎯 只有实际AI回复才保存
                            yield chunk
                            
                    except Exception as ai_error:
                        error_msg = f"\n\n❌ 生成回复失败: {str(ai_error)}"
                        actual_reply_content += error_msg  # 错误信息需要保存
                        yield error_msg
                
                # 🚀 发送消息ID信息
                message_info = f"<!--MESSAGE_ID:{system_msg.id}-->"
                yield message_info
                
            except Exception as stream_error:
                error_msg = f"\n\n❌ [系统异常] {str(stream_error)}"
                actual_reply_content += error_msg  # 异常信息需要保存
                yield error_msg
                logger.error(f"流式处理异常: {str(stream_error)}")
            
            finally:
                # 🎯 关键修改：只保存实际的AI回复内容，不包含进度信息
                try:
                    system_msg.message = actual_reply_content  # 只保存实际回复内容
                    db.add(system_msg)
                    await db.commit()
                    
                    # 更新相关任务的messageid字段
                  
                    from app.services.task_service import TaskService
                    updated_count = await TaskService.update_tasks_messageid_by_session(
                        session_id, str(system_msg.id), db, time_threshold_minutes=5
                    )
               
                except Exception as db_error:
                    logger.error(f"保存系统消息失败: {str(db_error)}")

        return StreamingResponse(optimized_stream(), media_type="text/event-stream")
        
    except Exception as e:
        # 对于异常情况，也以流式方式返回错误信息
        async def error_stream():
            error_msg = f"❌ 智能助手服务异常: {str(e)}\n\n请稍后重试，或尝试简化您的问题。"
            # 逐字符输出，提供更好的用户体验
            for char in error_msg:
                yield char
                await asyncio.sleep(0.01)
        
        return StreamingResponse(error_stream(), media_type="text/event-stream")


async def start_datafx_background(dataset_id: str, session: AsyncSession, table_structure: dict):
    """
    后台任务：执行数据分析并更新分析结果
    
    参数:
        dataset_id (str): 数据集ID
        session (AsyncSession): 数据库会话
        table_structure (dict): 数据集结构
    """
    # 创建一个新的会话，避免使用原有会话造成并发问题
    async with async_session_maker() as new_session:
        # 根据 sessionid 查询 UserSession 表以获取会话信息
        try:
         
            # 查询该会话已有的分析维度记录
            query = select(AnalysisDimension).where(AnalysisDimension.dataset_id == dataset_id)
            result = await new_session.execute(query)
            existing_dimensions = result.scalars().all()
                
            # 将已有的分析维度转换为字典格式，传递给大模型
            existing_dimensions_info = []
            for dim in existing_dimensions:
                dim_info = {
                    "分析维度": dim.dimension_name,
                    "分析描述": dim.dimension_description,
                    "分析类别": dim.analysis_goal,
                }
                existing_dimensions_info.append(dim_info)
            
            # 使用线程池执行大模型调用，避免阻塞事件循环，并传入已有的分析维度
            analysis_plans = await run_in_threadpool(
                advisor.generate_analysis_plan, 
                table_structure,
                existing_dimensions_info  # 传递已有的分析维度信息
            )
                        
            # 查询数据集信息以获取创建者ID
            dataset_query = select(DataSet).where(DataSet.id == dataset_id)
            dataset_result = await new_session.execute(dataset_query)
            dataset = dataset_result.scalar_one_or_none()
            creator_id = dataset.user_id if dataset else None
          
            # 添加新的分析维度记录（如果与维度同名，则跳过）
            for dimension in analysis_plans:
                dimension_name = dimension.get("分析维度", "")
                dimension_description = dimension.get("分析描述", "")
                analysis_goal = dimension.get("分析类别", "")
                
                # 创建新的分析维度记录
                new_dimension = AnalysisDimension(
                    dataset_id=dataset_id,
                    dimension_name=dimension_name,
                    dimension_description=dimension_description,
                    analysis_goal=analysis_goal,
                    answerable_questions="answerable_questions",
                    creator_id=creator_id,  # 添加创建人ID
                    created_at=datetime.now(),
                    is_locked=0  # 默认为未锁定状态
                )
                new_session.add(new_dimension)
                await new_session.commit()
                await new_session.refresh(new_dimension)
            
        except Exception as e:
            # 记录详细的错误信息
            logger.error(f"数据分析失败: {str(e)}")
            logger.exception(e)
            






@router.get("/get_user_sessions", response_model=List[UserSessionRead])
async def get_user_sessions(user: User = Depends(current_active_user), session: AsyncSession = Depends(get_async_session)):
    """
    获取当前用户的会话列表。
    """
    query = select(UserSession).where(
        UserSession.user_id == str(user.id)
    ).order_by(UserSession.upload_time.desc())
    result = await session.execute(query)
    conversations = result.scalars().all()
    return conversations


@router.post("/upload_file", response_model=dict)
async def upload_file(
    file: UploadFile = File(...),
    datasetname: str = Form(None),
    description: str = Form(None),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    上传文件并创建一个新的数据集。
    仅支持.xlsx和.xls格式文件。
    datasetname和description参数可选，为空时使用文件名(不含后缀)作为默认值。
    第一行应为列名，第二行开始为数据内容。
    """
    try:
        # 获取文件名（不含后缀）作为默认值
        file_stem = FilePath(file.filename).stem
        
        # 如果参数为空，使用文件名作为默认值
        if not datasetname:
            datasetname = file_stem
        
        if not description:
            description = file_stem
            
        # 检查文件扩展名
        file_extension = FilePath(file.filename).suffix.lower()
        if file_extension not in ['.xlsx', '.xls']:
            raise FileUploadException(
                message="文件格式不支持",
                detail=f"只支持.xlsx和.xls格式的Excel文件，当前文件格式为: {file_extension}"
            )

        upload_dir = datetime.now().strftime("%Y/%m")
        file_name = f"{uuid.uuid4()}{file_extension}"
        file_path = os.path.join(upload_dir, file_name)

        # 确保目录存在
        os.makedirs(upload_dir, exist_ok=True)

        # 保存文件
        try:
            with open(file_path, "wb") as f:
                f.write(await file.read())
        except Exception as e:
            raise FileUploadException(
                message="保存文件失败",
                detail=f"文件保存过程中出错: {str(e)}"
            )

        # 使用pandas读取Excel文件
        try:
            # 先读取前几行进行结构验证
            df_preview = pd.read_excel(file_path, nrows=3)
            
            # 检查是否至少有两行数据（标题行和数据行）
            if len(df_preview) < 2:
                raise FileUploadException(
                    message="Excel文件格式不正确",
                    detail="文件必须至少包含两行：第一行为列名，第二行开始为数据"
                )
            
            # 检查第一行是否可能是列名（不包含缺失值）
            if df_preview.iloc[0].isna().any():
                logger.warning("检测到第一行包含空值，可能不是标准的列名行")
                
            # 读取完整数据
            df = DataHelp.read_excel_file(file_path)
            if df.empty:
                raise FileUploadException(
                    message="上传的Excel文件为空",
                    detail="文件不包含任何数据，请检查文件内容"
                )
        except Exception as e:
            raise FileUploadException(
                message="读取Excel文件失败",
                detail=f"解析文件时出错: {str(e)}"
            )

        try:
            datacle = EnhancedDataCleaner()
            cleaned_df, features, clean_log = datacle.process_data(df)
            data_summary = datacle.generate_data_summary(cleaned_df)
        except Exception as e:
            raise DataAnalysisException(
                message="数据处理失败",
                detail=f"清洗数据时出错: {str(e)}"
            )
         
        # 创建一个新的DataSet实例
        new_dataSet = DataSet(
            user_id=str(current_user.id),
            description=description,
            data_source_id='00000000-0000-0000-0000-000000000000',
            name=datasetname or FilePath(file.filename).stem,  # 使用提供的名称或文件名
            type='xls',
            custom_query=file_path,
            cleaning_log=clean_log,
            data_summary_log=DataHelp.to_json(data_summary),
            created_at=datetime.now()
        )
        
        try:
            session.add(new_dataSet)
            await session.flush()  # 提交以获取数据集ID
            await session.refresh(new_dataSet)
            
            table_name = DataHelp.get_table_name(new_dataSet.id)            
            
        except SQLAlchemyError as e:
            await session.rollback()
            raise DatabaseException(
                message="保存数据集数据失败",
                detail=f"数据库操作错误: {str(e)}"
            )

        try:
            # 保存数据到MySQL
            cleaned_df.to_sql(table_name, sync_engine,
                          index=False, if_exists='replace')
            new_dataSet.original_name = table_name
            
            # 步骤1: 首先创建字段记录
            await DataHelp.create_dataset_fields(cleaned_df, new_dataSet.id, session)
            
            # 步骤2: 然后获取数据集结构信息
            dataset_structure = await DataHelp.get_dataset_structure(new_dataSet.id, session)
            new_dataSet.dataset_summary = DataHelp.to_json(dataset_structure)
            
            await session.commit()
        except Exception as e:
            await session.rollback()
            raise DatabaseException(
                message="创建数据表失败",
                detail=f"保存数据到数据库时出错: {str(e)}"
            )
            
        return new_dataSet.to_dict()
        
    except (FileUploadException, DataAnalysisException, DatabaseException):
        # 这些异常会被全局异常处理器捕获并处理
        raise
    except Exception as e:
        # 处理未预期的异常
        logger.error(f"上传文件时发生未预期的错误: {str(e)}")
        logger.exception(e)
        from app.utils.error_handler import APIException, ErrorCode
        raise APIException(
            code=ErrorCode.SYSTEM_ERROR,
            message="上传文件过程中发生系统错误",
            detail=str(e)
        )

@router.post("/start_analysis", response_model=dict)
async def start_analysis(
    background_tasks: BackgroundTasks,
    dataset_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    通过数据集ID开始生成推荐分析维度。

    """
   
            # 查询数据集
    dataset_query = select(DataSet).where(DataSet.id == dataset_id)
    dataset_result = await session.execute(dataset_query)
    dataset = dataset_result.scalar_one_or_none()
    
    if dataset and dataset.dataset_summary:
        # 使用数据集的dataset_summary字段
        table_info = json.loads(dataset.dataset_summary)
    
    # 将后台任务添加到 BackgroundTasks
    # background_tasks.add_task(start_datafx_background, sessionid, session, df)
    await start_datafx_background(dataset_id, session, table_info)

    return {"status": "success"}


@router.post("/analysis_status", response_model=dict)
async def analysis_status(
    sessionid: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    获取指定会话的分析状态和分析结果。
    分析结果从 analysis_dimensions 表中获取，按照维度进行组织。
    """
    # 查询会话信息
    query = select(UserSession).where(UserSession.id == sessionid)
    result = await session.execute(query)
    user_session = result.scalar()

    if not user_session:
        raise HTTPException(status_code=404, detail="会话出错")
    
    # 查询关联的分析维度
    dimensions_query = select(AnalysisDimension).where(AnalysisDimension.session_id == sessionid)
    dimensions_result = await session.execute(dimensions_query)
    dimensions = dimensions_result.scalars().all()
    
    # 将分析维度转换为适合前端的格式
    analysis_dimensions = []
    for dimension in dimensions:
        dimension_dict = dimension.dict()
        
        # 处理图表建议，尝试将JSON字符串转换为Python对象
        if dimension.chart_suggestions:
            try:
                dimension_dict["chart_suggestions"] = json.loads(dimension.chart_suggestions)
            except:
                pass
        
        analysis_dimensions.append(dimension_dict)
    
    # 将会话信息和分析维度组合成一个字典
    session_details = {
        "analysis_status": user_session.analysis_status,
        "analysis_dimensions": analysis_dimensions,  # 新增字段，包含从表中查询的分析维度
        "fxfxstatus": user_session.analysis_status
    }

    return session_details


@router.post("/get_session_details", response_model=dict)
async def get_session_details(
    session_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    获取指定会话的详细信息以及该会话内的历史消息。
    还会返回该会话关联的分析维度信息。
    """

    # 根据 sessionid 查询 UserSession 表以获取会话信息
    query = select(UserSession).where(UserSession.id == session_id)
    result = await session.execute(query)
    user_session = result.scalar()
    if not user_session:
        raise HTTPException(status_code=404, detail="会话出错")

   
   

   
    # 获取会话关联的第一个数据集信息
    dataset_info = None
    dimensions = []
    if user_session.dataset_ids:
        dataset_id_list = user_session.get_dataset_ids()
        if dataset_id_list:
            # 只查询第一个关联的数据集
            first_dataset_id = dataset_id_list[0]
            dataset_query = select(DataSet).where(DataSet.id == first_dataset_id)
            dataset_result = await session.execute(dataset_query)
            dataset = dataset_result.scalar_one_or_none()
            
            if dataset:
                dataset_info = dataset.to_dict()
            # 查询关联的分析维度
            dimensions_query = select(AnalysisDimension).where(AnalysisDimension.dataset_id == dataset.id,AnalysisDimension.dimension_type=='user_dimension')
            dimensions_result = await session.execute(dimensions_query)
            dimensions = dimensions_result.scalars().all()



    
    # 将会话信息和历史消息组合成一个字典
    session_details = {
        "analysis_status": user_session.analysis_status,
        "analysis_dimensions": dimensions,  # 新增字段，包含从表中查询的分析维度
        "dataset": dataset_info
    }

    return session_details


@router.post("/send_msg", response_model=UserMsgRead)
async def send_msg(
    message: str = Form(...),
    session_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    发送消息并获取大模型回复
    """
    try:

        query = select(UserSession).where(UserSession.id == session_id)
        result = await session.execute(query)
        user_session = result.scalar()
        if not user_session:
            raise HTTPException(status_code=404, detail="调用出错")
        # 创建一个新的 UserMsg 实例
        new_msg = UserMsg(
            user_id=str(user.id),
            session_id=session_id,
            message=message,
            message_type='user',
            timestamp=datetime.now()
        )
        # 添加到会话并提交
        session.add(new_msg)
        await session.commit()
        await session.refresh(new_msg)
        return new_msg

    except Exception as e:
        await session.rollback()
        raise HTTPException(status_code=500, detail=str(e))


@router.post("/delete_session", response_model=dict)
async def delete_session(
    sessionid: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据会话ID删除会话及其关联的消息。
    """
    # 根据 sessionid 查询 UserSession 表以获取会话信息
    query = select(UserSession).where(UserSession.id == sessionid)
    result = await session.execute(query)
    user_session = result.scalar()

    if not user_session:
        raise HTTPException(status_code=404, detail="会话出错")

    # 删除所有与该会话关联的 UserMsg 记录
    query = select(UserMsg).where(UserMsg.session_id == sessionid)
    result = await session.execute(query)
    messages = result.scalars().all()
    for msg in messages:
        await session.delete(msg)

    # 删除会话
    await session.delete(user_session)
    await session.commit()

    table_name = DataHelp.get_table_name(sessionid)

    drop_table(table_name)
  
    return {"status": "success", "message": "会话及其关联的消息已成功删除"}


@router.post("/delete_message", response_model=dict)
async def delete_message(
    message_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据消息ID删除消息。
    """
    # 根据 message_id 查询 UserMsg 表以获取消息信息
    query = select(UserMsg).where(UserMsg.id == message_id)
    result = await session.execute(query)
    user_msg = result.scalar()

    if not user_msg:
        raise HTTPException(status_code=404, detail="消息不存在")

    # 删除消息
    await session.delete(user_msg)
    await session.commit()

    return {"status": "success", "message": "消息已成功删除"}


@router.post("/get_msg_fx", response_model=None)
async def get_msg_fx(
    message_id: str = Form(...),
    umessage_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    流式输出大模型的反馈内容，并将流式消息存储到数据库中的UserMsg表中。
    数据来源从AnalysisDimension表中的任务数据获取。
    """
    try:
        # 查询用户消息
        uquery = select(UserMsg).where(UserMsg.id == umessage_id)
        uresult = await session.execute(uquery)
        uuser_msg = uresult.scalar()
        
        if not uuser_msg:
            raise HTTPException(status_code=404, detail="用户消息不存在")

        # 查询系统消息
        query = select(UserMsg).where(UserMsg.id == message_id)
        result = await session.execute(query)
        user_msg = result.scalar()
        
        if not user_msg:
            raise HTTPException(status_code=404, detail="系统消息不存在")

        # 从AnalysisDimension表中获取任务数据
        # 查询与该消息关联的已完成或失败的数据查询任务
        tasks_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == umessage_id,
            AnalysisDimension.dimension_type == "task",
            AnalysisDimension.analysis_method == "数据查询",
            AnalysisDimension.status.in_(["completed", "failed"])
        )
        tasks_result = await session.execute(tasks_query)
        data_query_tasks = tasks_result.scalars().all()
        
        # 收集已完成或失败的数据查询任务
        candidate_tasks = []
        
        for task in data_query_tasks:
            # 解析任务要求以获取数据说明
            task_requirements = {}
            datadec = ""
            try:
                if task.dimension_description:
                    task_requirements = json.loads(task.dimension_description)
                    datadec = task_requirements.get("说明", "")
            except:
                datadec = task.dimension_description or ""
            
            # 根据任务状态构建任务数据对象
            if task.status == "completed":
                # 成功完成的任务，包含查询结果数据
                if task.json_data:
                    try:
                        task_data = json.loads(task.json_data)
                        task_data_item = {
                            "task_name": task.dimension_name,
                            "task_id": task.id,
                            "task_type": "数据查询",
                            "status": "completed",
                            "datadec": datadec,
                            "data": task_data,
                            "data_count": len(task_data) if isinstance(task_data, list) else 0,
                            "message": "任务执行成功"
                        }
                    except Exception as e:
                        # JSON解析失败的情况
                        task_data_item = {
                            "task_name": task.dimension_name,
                            "task_id": task.id,
                            "task_type": "数据查询",
                            "status": "error",
                            "datadec": datadec,
                            "data": [],
                            "data_count": 0,
                            "error_message": f"数据解析失败: {str(e)}",
                            "message": "任务数据格式错误"
                        }
                else:
                    # 已完成但无数据的任务
                    task_data_item = {
                        "task_name": task.dimension_name,
                        "task_id": task.id,
                        "task_type": "数据查询",
                        "status": "completed",
                        "datadec": datadec,
                        "data": [],
                        "data_count": 0,
                        "message": "任务执行成功但无数据返回"
                    }
            elif task.status == "failed":
                # 失败的任务，从analysis_result字段获取失败原因
                error_message = task.analysis_result or "数据查询任务执行失败，未知原因"
                task_data_item = {
                    "task_name": task.dimension_name,
                    "task_id": task.id,
                    "task_type": "数据查询",
                    "status": "failed",
                    "datadec": datadec,
                    "data": [],
                    "data_count": 0,
                    "error_message": error_message,
                    "message": "数据查询任务执行失败"
                }
            
            candidate_tasks.append(task_data_item)
        
        # 按任务状态和数据量排序，优先处理失败的任务，然后是成功的任务（按数据量排序）
        def sort_key(task):
            # 失败的任务优先级最高（返回值最小）
            if task["status"] == "failed":
                return (0, 0)
            # 数据解析错误的任务次优先级
            elif task["status"] == "error":
                return (1, 0)
            # 成功的任务按数据量排序，数据量小的优先
            elif task["status"] == "completed":
                return (2, task["data_count"])
            # 其他状态的任务
            else:
                return (3, 0)
        
        candidate_tasks.sort(key=sort_key)
        
        # 检查candidate_tasks是否超过提示词长度限制
        try:
            # 将candidate_tasks转换为字符串进行token检查
            tasks_json_str = json.dumps(candidate_tasks, ensure_ascii=False)
            user_message_str = uuser_msg.message or ""
            
            # 组合用户消息和任务数据，模拟完整的提示词
            combined_prompt = f"用户问题: {user_message_str}\n\n任务数据: {tasks_json_str}"
            
            within_limit, token_info = check_prompt_limit(combined_prompt)
            
            if not within_limit:
                # 如果超过限制，生成包含任务状态统计的错误提示信息
                completed_count = len([t for t in candidate_tasks if t["status"] == "completed"])
                failed_count = len([t for t in candidate_tasks if t["status"] == "failed"])
                error_count = len([t for t in candidate_tasks if t["status"] == "error"])
                
                error_msg = f"""数据查询任务信息量过大，无法完整分析。任务执行统计：
                            - 查询成功：{completed_count}个任务
                            - 查询失败：{failed_count}个任务  
                            - 数据错误：{error_count}个任务
                            - 总任务数：{len(candidate_tasks)}个
                            建议：请缩小查询范围或分批处理任务。"""
                
                # 以流式方式返回错误提示
                async def stream_error_message():
                    # 将错误信息分块流式输出
                    error_chunks = error_msg.split('\n')
                    full_message = ""
                    
                    for chunk_text in error_chunks:
                        chunk_with_newline = chunk_text + '\n'
                        full_message += chunk_with_newline
                        yield chunk_with_newline
                        # 添加小延迟模拟流式输出效果
                        await asyncio.sleep(0.05)
                    
                    # 更新系统消息内容
                    user_msg.message = full_message.strip()
                    session.add(user_msg)
                    await session.commit()
                
                return StreamingResponse(stream_error_message(), media_type="text/event-stream")
                
            # 统计任务状态
            completed_count = len([t for t in candidate_tasks if t["status"] == "completed"])
            failed_count = len([t for t in candidate_tasks if t["status"] == "failed"])
            error_count = len([t for t in candidate_tasks if t["status"] == "error"])
                    
        except Exception as token_check_error:
            logger.error(f"Token检查失败: {str(token_check_error)}")
            # Token检查失败时，记录错误但继续执行
        
        async def store_and_stream():
            def generate_sync():
                # 使用从任务中获取的数据进行分析
                yield from advisor.generate_hfmsg(
                    uuser_msg.message,
                    candidate_tasks
                )

            # 包装为异步生成器
            sysmsg = ""
            for chunk in generate_sync():
                # 存储每一块流式消息到数据库
                sysmsg = sysmsg + chunk
                yield chunk
            
            # 更新系统消息内容
            user_msg.message = sysmsg
            session.add(user_msg)
            await session.commit()

        return StreamingResponse(store_and_stream(), media_type="text/event-stream")
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"处理失败: {str(e)}")


@router.post("/get_msg", response_model=None)
async def get_msg(
    message_id: str = Form(...),
    session_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    返回SQL执行结果和流式输出大模型的反馈内容，并将流式消息存储到数据库中的UserMsg表中。
    SQL结果会作为第一条完整消息返回，随后是大模型的流式响应。
    支持处理多条SQL语句(以分号;分隔)，并将所有SQL执行结果存储。
    """
    try:
        # 查询用户消息
        query = select(UserMsg).where(UserMsg.id == message_id)
        result = await session.execute(query)
        user_msg = result.scalar()
        if not user_msg:
            raise HTTPException(status_code=404, detail="消息不存在")

        # 查询会话信息
        squery = select(UserSession).where(UserSession.id == session_id)
        sresult = await session.execute(squery)
        user_session = sresult.scalar()
        if not user_session:
            raise HTTPException(status_code=404, detail="会话出错")
        # 计算5分钟前的时间
        five_minutes_ago = datetime.now() - timedelta(minutes=5)

        # 查询最近3条时间不超过5分钟的会话信息
        query = select(UserMsg).where(
            UserMsg.session_id == session_id, 
            UserMsg.message_type == 'user',
            UserMsg.timestamp >= five_minutes_ago
        ).order_by(UserMsg.timestamp.desc()).limit(3)
        result = await session.execute(query)
        user_msgs = result.scalars().all()
        
        # 将 user_msgs 的 message 字段用逗号拼接
        user_msgs_json = ",".join([msg.message for msg in user_msgs])

        
        # 从会话关联的数据集获取table_info
        table_info = None
        if user_session.dataset_ids:
            # 获取第一个数据集ID
            dataset_id_list = user_session.get_dataset_ids()
            if dataset_id_list:
                first_dataset_id = dataset_id_list[0]
                # 查询数据集
                dataset_query = select(DataSet).where(DataSet.id == first_dataset_id)
                dataset_result = await session.execute(dataset_query)
                dataset = dataset_result.scalar_one_or_none()
                
                if dataset and dataset.dataset_summary:
                    # 使用数据集的dataset_summary字段
                    table_info = json.loads(dataset.dataset_summary)
        
        # 调用大模型生成SQL
        responseai = await run_in_threadpool(advisor.generate_sql_plan,table_info, user_msg.message,user_msgs_json)
        logger.debug('用户消息:'+user_msg.message+'大模型回复:' + DataHelp.to_json(responseai))
        
        if not responseai or len(responseai) == 0:
            raise HTTPException(status_code=400, detail="大模型返回结果为空")

        useryt = responseai[0]["用户主要意图"]
        userrequr = responseai[0]["需求确认信息"]
        userqustion = responseai[0]["用户其它意图"]
        
        # 获取数据源信息中的SQL说明，可能有多个
        data_sources = responseai[0].get("数据源信息", [])
        sql_remarks = []
        for source in data_sources:
            if "SQL说明" in source and source["SQL说明"]:
                sql_remark = f"{source.get('数据源名称', '未命名数据源')}: {source['SQL说明']}"
                sql_remarks.append(sql_remark)
        
        # 合并所有SQL说明
        message_remark = "\n".join(sql_remarks) if sql_remarks else ""
        
        # 更新用户消息的相关字段
        user_msg.message_intent = useryt
        user_msg.message_confirm = userrequr
        user_msg.message_suggest = userqustion
        user_msg.message_remark = message_remark  # 添加合并后的SQL说明到message_remark字段
        await session.commit()
        
        # 获取数据源信息中的SQL语句
        data_sources = responseai[0].get("数据源信息", [])
        if not data_sources:
            raise HTTPException(status_code=400, detail="未找到数据源信息")
            
        # 分割SQL语句并执行每一条
        all_query_results = []
        all_sql_statements = []
        
       
                
           
        
        # 处理所有数据源的SQL语句
        for source in data_sources:
            sql_statement = source.get("SQL", "").strip()
            if not sql_statement:
                continue
                
            logger.debug(f"执行SQL语句: {sql_statement}")
            all_sql_statements.append(sql_statement)
            
            try:
                result = await DatasetService.execute_sql_from_source(dataset.data_source_id, sql_statement,session)

                result_with_meta = {
                    "data": result["data"],
                    "数据源名称": source.get("数据源名称", ""),
                    "推荐图表类型": source.get("推荐图表类型", "")
                }
                all_query_results.append(result_with_meta)
                
            except Exception as e:
                logger.error(f"执行SQL出错: {str(e)}")
                all_query_results.append({
                    "error": str(e),
                    "数据源名称": source.get("数据源名称", ""),
                    "推荐图表类型": source.get("推荐图表类型", "")
                })
        
        if not all_query_results:
            raise HTTPException(status_code=400, detail="所有SQL语句执行失败")        
        # 创建系统消息记录
        new_msg = UserMsg(
            user_id=str(user.id),
            session_id=session_id,
            message='系统回复',
            message_type='system',
            message_sourceid=message_id,
            message_sql=DataHelp.to_json(all_sql_statements),  # 存储所有SQL语句
            message_data=DataHelp.to_json(all_query_results),  # 存储所有查询结果
            message_intent=useryt,
            message_confirm=userrequr,
            message_suggest=userqustion,
            message_remark=message_remark,  # 添加合并后的SQL说明到message_remark字段
            timestamp=datetime.now()
        )
        session.add(new_msg)
        await session.commit()
        
        return {
            "status": "success",
            "dimensions": DataHelp.to_json(all_query_results),
            "hfmsgid": new_msg.id,
            "hfmsgintent": new_msg.message_intent

        }
        
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        logger.error(f"处理消息时发生错误: {str(e)}")
        raise HTTPException(status_code=500, detail=f"处理消息失败: {str(e)}")


@router.get("/get_analysis_dimensions", response_model=List[dict])
async def get_analysis_dimensions(
    session_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    获取指定会话的所有分析维度信息
    
    参数:
        session_id (str): 会话ID
        
    返回:
        List[dict]: 分析维度信息列表
    """
    try:
        # 查询该会话的所有分析维度
        query = select(AnalysisDimension).where(AnalysisDimension.session_id == session_id)
        result = await session.execute(query)
        dimensions = result.scalars().all()
        
        # 将查询结果转换为字典列表
        dimension_list = []
        for dimension in dimensions:
            dimension_dict = dimension.dict()
            
            # 只对图表建议进行 JSON 解析，analysis_sql 和 answerable_questions 保持字符串形式
            if dimension.chart_suggestions:
                try:
                    dimension_dict["chart_suggestions"] = json.loads(dimension.chart_suggestions)
                except:
                    pass
                    
            dimension_list.append(dimension_dict)
            
        return dimension_list
        
    except Exception as e:
        logger.error(f"获取分析维度信息失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取分析维度信息失败: {str(e)}")

@router.post("/get_analysis_dimension", response_model=dict)
async def get_analysis_dimension_by_id(
    dimension_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据ID获取单个分析维度的详细信息
    
    参数:
        dimension_id (str): 分析维度ID
        
    返回:
        dict: 分析维度详细信息
    """
    try:
        # 查询指定ID的分析维度
        query = select(AnalysisDimension).where(AnalysisDimension.id == dimension_id)
        result = await session.execute(query)
        dimension = result.scalar_one_or_none()
        
        if not dimension:
            raise HTTPException(status_code=404, detail=f"分析维度不存在: {dimension_id}")
        
        # 转换为字典并解析JSON字段
        dimension_dict = dimension.dict()
        
        # 只对图表建议进行 JSON 解析，analysis_sql 和 answerable_questions 保持字符串形式
        if dimension.chart_suggestions:
            try:
                dimension_dict["chart_suggestions"] = json.loads(dimension.chart_suggestions)
            except:
                pass
        
        return {
            "status": "success",
            "data": dimension_dict
        }        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取分析维度信息失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取分析维度信息失败: {str(e)}")

@router.post("/toggle_dimension_lock", response_model=None)
async def toggle_dimension_lock(
    dimension_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    切换分析维度的锁定状态
    
    参数:
        dimension_id (str): 分析维度ID
        
    返回:
        dict: 包含新状态的响应
    """
    try:
        # 查询指定ID的分析维度
        query = select(AnalysisDimension).where(AnalysisDimension.id == dimension_id)
        result = await session.execute(query)
        dimension = result.scalar_one_or_none()
        
        if not dimension:
            raise HTTPException(status_code=404, detail=f"分析维度不存在: {dimension_id}")
        
        # 切换锁定状态
        dimension.is_locked = 0 if dimension.is_locked else 1
        
        # 保存更改
        session.add(dimension)
        await session.commit()
        
        return {
            "status": "success",
            "message": f"分析维度 '{dimension.dimension_name}' 的锁定状态已更新",
            "dimension_id": dimension_id,
            "is_locked": bool(dimension.is_locked)
        }
        
    except HTTPException:
        raise
    except Exception as e:
        await session.rollback()
        logger.error(f"更新分析维度锁定状态失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"更新分析维度锁定状态失败: {str(e)}")

@router.post("/generate_dimension_charts", response_model=None)
async def generate_dimension_charts(
    message_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据消息ID生成Echarts图表配置
    
    参数:
        message_id (str): 用户消息ID
        
    返回:
        dict: 包含Echarts配置项的响应
    """
    try:
        # 查询指定ID的消息
        query = select(UserMsg).where(UserMsg.id == message_id)
        result = await session.execute(query)
        message = result.scalar_one_or_none()
        
        if not message:
            raise HTTPException(status_code=404, detail=f"消息不存在: {message_id}")
        
        # 查询会话信息
        session_id = message.session_id
        query = select(UserSession).where(UserSession.id == session_id)
        result = await session.execute(query)
        user_session = result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(status_code=404, detail=f"会话不存在: {session_id}")
        
        # 获取表名
        table_name = DataHelp.get_table_name(session_id)
        
        # 从消息中提取SQL语句
        all_sql_statements = []
        if message.message_sql:
            try:
                all_sql_statements = json.loads(message.message_sql)
            except Exception as e:
                logger.error(f"解析消息中的SQL失败: {str(e)}")
                all_sql_statements = [message.message_sql]  # 如果解析失败，将整个内容作为一条SQL语句
        
        # 从消息中提取数据
        all_query_results = []
        if message.message_data:
            try:
                all_query_results = json.loads(message.message_data)
            except Exception as e:
                logger.error(f"解析消息中的数据失败: {str(e)}")
                raise HTTPException(status_code=400, detail=f"消息数据格式不正确: {str(e)}")
        
        if not all_query_results:
            raise HTTPException(status_code=400, detail="消息中没有可用的查询结果数据")
        
        # 准备图表生成所需的数据结构
        chart_data = []
        
        # 为每个数据源和SQL构建图表配置
        for i, result_data in enumerate(all_query_results):
            # 提取数据源信息
            data_source_name = result_data.get("数据源名称", f"数据源{i+1}")
            chart_type = result_data.get("推荐图表类型", "auto")
            
            # 获取SQL语句
            sql = all_sql_statements[i] if i < len(all_sql_statements) else ""
            
            # 获取实际数据
            data = result_data.get("data", [])
            
            # 准备查询结果JSON字符串
            data_json = DataHelp.to_json(data)
            
            # 调用generate_echarts_config生成图表配置
            try:
                # 查询源消息
                source_message = None
                if message.message_sourceid:
                    source_query = select(UserMsg).where(UserMsg.id == message.message_sourceid)
                    source_result = await session.execute(source_query)
                    source_message = source_result.scalar_one_or_none()
                
                # 使用源消息的内容，如果没有找到则使用当前消息内容
                user_request = source_message.message if source_message else message.message
                
                # echarts_config = await run_in_threadpool(
                #     advisor.generate_echarts_config,
                #     sql,
                #     chart_type,
                #     user_request,  # 使用源消息的原始请求
                #     data_json  # 实际数据
                # )
                
                g2_config = await run_in_threadpool(
                    advisor.generate_g2_config,
                    sql,
                    chart_type,
                    user_request,  # 使用源消息的原始请求
                    data_json  # 实际数据
                )
                # 将结果添加到图表数据中
                chart_data.append({
                    "数据源名称": data_source_name,
                    "推荐图表类型": chart_type,
                    "SQL": sql,
                    "echarts_config": "",
                    "g2_config": g2_config
                })
                
            except Exception as e:
                logger.error(f"为数据源 '{data_source_name}' 生成Echarts配置失败: {str(e)}")
                chart_data.append({
                    "数据源名称": data_source_name,
                    "推荐图表类型": chart_type,
                    "SQL": sql,
                    "error": str(e)
                })
        
        # 更新消息的chart_option字段
        message.chart_option = DataHelp.to_json(chart_data)
        session.add(message)
        await session.commit()
        
        # 返回Echarts配置
        return {
            "status": "success",
            "message_id": message_id,
            "session_id": session_id,
            "chart_data": chart_data
        }
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"生成Echarts图表配置失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"生成Echarts图表配置失败: {str(e)}")

@router.post("/generate_report", response_model=None)
async def generate_report(
    report_type: str = Form(...),
    session_id: str = Form(...),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据报告类型和会话ID生成对应的报告，流式输出
    
    参数:
        report_type (int): 报告类型 cleaning:清洗报告 summary:摘要报告
        session_id (str): 会话ID
        
    注意:
        如需生成分析大纲，请使用专门的/generate_analysis_outline接口
        
    返回:
        StreamingResponse: 流式输出的报告内容
    """
    try:
        # 查询会话信息
        query = select(UserSession).where(UserSession.id == session_id)
        result = await session.execute(query)
        user_session = result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(status_code=404, detail=f"会话不存在: {session_id}")
        
        
        # 检查会话是否关联了数据集
        if not user_session.dataset_ids:
            raise HTTPException(status_code=400, detail="该会话没有关联的数据集")
        
        # 获取第一个数据集ID
        dataset_id_list = user_session.get_dataset_ids()
        if not dataset_id_list:
            raise HTTPException(status_code=400, detail="无法获取关联的数据集ID")
        
        # 查询数据集信息
        first_dataset_id = dataset_id_list[0]
        dataset_query = select(DataSet).where(DataSet.id == first_dataset_id)
        dataset_result = await session.execute(dataset_query)
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(status_code=404, detail=f"关联的数据集不存在: {first_dataset_id}")
        # 根据报告类型选择不同的处理逻辑和标题
        if report_type == "cleaning":  # 清洗报告
          
            
            # 判断是否为本地数据集
            if dataset.data_source_id != '00000000-0000-0000-0000-000000000000':
                raise HTTPException(status_code=400, detail="非本地数据集没有清洗报告")
            # 从数据集的cleaning_log字段获取清洗日志
            if not dataset.cleaning_log:
                raise HTTPException(status_code=400, detail="该数据集没有清洗日志")
            
            report_data = dataset.cleaning_log
            report_title = "数据清洗优化报告"
            
        elif report_type == "summary":  # 摘要报告
            report_data = await DataHelp.generate_dataset_summary(first_dataset_id, session)
            report_title = "数据概览报告"
        else:
            raise HTTPException(status_code=400, detail=f"不支持的报告类型: {report_type}")
        
        if report_data is None:
            raise HTTPException(status_code=400, detail="报告数据为空")
        
        # 使用AIGetData的方法流式生成报告
        return StreamingResponse(
            advisor.generate_report(report_type, report_data, report_title),
            media_type="text/event-stream"
        )
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"生成报告失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"生成报告失败: {str(e)}")

@router.post("/mydatasets", response_model=None)
async def get_my_datasets(
    search: str = Form(""),
    dataset_type: int = Form(0, description="数据集类型: 0=全部数据集, 1=本地数据集, 2=非本地数据集"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """获取当前用户有权限的数据集
    
    返回的数据集包括:
    1. 用户创建的数据集
    2. 用户角色被授权的数据集
    
    Args:
        search: 搜索关键词
        dataset_type: 数据集类型: 0=全部数据集(默认), 1=本地数据集, 2=非本地数据集
    """
    try:
        # 获取用户角色
        user_roles = await get_user_role_codes(str(current_user.id), session)
        
        # 1. 先获取用户自己创建的数据集
        created_query = select(DataSet).where(DataSet.user_id == str(current_user.id))
        
        # 根据数据集类型筛选
        if dataset_type == 1:  # 本地数据集
            created_query = created_query.filter(DataSet.data_source_id == '00000000-0000-0000-0000-000000000000')
        elif dataset_type == 2:  # 非本地数据集
            created_query = created_query.filter(DataSet.data_source_id != '00000000-0000-0000-0000-000000000000')
        
        # 应用搜索条件
        if search and search.strip():
            created_query = created_query.filter(
                or_(
                    DataSet.name.ilike(f"%{search}%"),
                    DataSet.description.ilike(f"%{search}%"),
                    DataSet.original_name.ilike(f"%{search}%")
                )
            )
        
        result = await session.execute(created_query)
        created_datasets = result.scalars().all()
        
        # 2. 获取用户通过角色访问的数据集
        role_names = user_roles.split(",") if user_roles else []
        authorized_datasets = []
        
        if role_names:
            # 获取角色ID，使用distinct()去重
            roles_result = await session.execute(
                select(Role.id).distinct().where(Role.name.in_(role_names))
            )
            role_ids = [str(r[0]) for r in roles_result.all()]
            
            if role_ids:
                # 查询授权数据集，使用distinct()去重
                current_time = datetime.now()
                authorized_query = select(DataSet).distinct().join(
                    RoleDatasetAccess,
                    DataSet.id == RoleDatasetAccess.dataset_id
                ).where(
                    RoleDatasetAccess.role_id.in_(role_ids),
                    or_(
                        RoleDatasetAccess.expires_at == None,
                        RoleDatasetAccess.expires_at > current_time
                    )
                )
                
                # 根据数据集类型筛选
                if dataset_type == 1:  # 本地数据集
                    authorized_query = authorized_query.filter(DataSet.data_source_id == '00000000-0000-0000-0000-000000000000')
                elif dataset_type == 2:  # 非本地数据集
                    authorized_query = authorized_query.filter(DataSet.data_source_id != '00000000-0000-0000-0000-000000000000')
                
                # 应用相同的搜索条件
                if search and search.strip():
                    authorized_query = authorized_query.filter(
                        or_(
                            DataSet.name.ilike(f"%{search}%"),
                            DataSet.description.ilike(f"%{search}%"),
                            DataSet.original_name.ilike(f"%{search}%")
                        )
                    )
                
                result = await session.execute(authorized_query)
                authorized_datasets = result.scalars().all()
        
        # 3. 合并结果并去重（在Python中完成）
        dataset_dict = {}
        for ds in created_datasets:
            dataset_dict[ds.id] = ds
        
        for ds in authorized_datasets:
            if ds.id not in dataset_dict:  # 避免重复
                dataset_dict[ds.id] = ds
        
        # 4. 按创建时间排序
        all_datasets = list(dataset_dict.values())
        all_datasets.sort(key=lambda x: x.created_at, reverse=True)
        
        # 构建返回结果
        return {
            "success": True,
            "data": [ds.to_dict() for ds in all_datasets],
            "user_roles": user_roles
        }
    
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"获取用户权限数据集失败: {str(e)}"
        )

@router.post("/dataset-data/{dataset_id}", response_model=None)
async def get_dataset_data(
    dataset_id: str = Path(..., title="数据集ID"),
    page: int = Form(1, ge=1, description="页码，从1开始"),
    page_size: int = Form(50, ge=1, le=1000, description="每页记录数"),
    search: str = Form(...),
    filter_visible: bool = Form(True, description="是否只显示标记为可见的字段"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据数据集ID获取对应的数据，支持分页
    
    Args:
        dataset_id: 数据集ID
        page: 页码，从1开始
        page_size: 每页记录数
        search: 搜索关键词
        filter_visible: 是否只返回标记为可见的字段，默认为True
    """
    try:
        # 1. 权限验证 - 检查用户是否有权限访问该数据集
        user_roles = await get_user_role_codes(str(current_user.id), session)

        # 先检查是否是用户自己创建的数据集
        dataset_query = select(DataSet).where(
            DataSet.id == dataset_id,
            DataSet.user_id == str(current_user.id)
        )
        result = await session.execute(dataset_query)
        dataset_row = result.first()  # 返回的是Row对象
        dataset = dataset_row[0] if dataset_row else None  # 获取Row中的DataSet对象
        
        # 如果不是自己创建的，检查是否通过角色授权
        if dataset is None:
            # 获取用户的角色ID
            role_names = user_roles.split(",") if user_roles else []
            role_ids = []
            if role_names:
                # 添加distinct()去重
                roles_result = await session.execute(
                    select(Role.id).distinct().where(Role.name.in_(role_names))
                )
                role_ids = [str(r[0]) for r in roles_result.all()]
            
            # 如果用户有角色，检查是否被授权访问该数据集
            if role_ids:
                current_time = datetime.now()
                # 添加distinct()去重
                access_query = select(DataSet).distinct().join(
                    RoleDatasetAccess,
                    DataSet.id == RoleDatasetAccess.dataset_id
                ).where(
                    DataSet.id == dataset_id,
                    RoleDatasetAccess.role_id.in_(role_ids),
                    or_(
                        RoleDatasetAccess.expires_at == None,
                        RoleDatasetAccess.expires_at > current_time
                    )
                )
                result = await session.execute(access_query)
                dataset_row = result.first()  # 返回的是Row对象
                dataset = dataset_row[0] if dataset_row else None  # 获取Row中的DataSet对象
            
            # 如果仍然没有找到数据集，用户没有权限
            if dataset is None:
                raise HTTPException(
                    status_code=403,
                    detail="您没有权限访问此数据集"
                )
        
        # 1.5 获取可见字段列表（如果需要过滤）
        visible_fields = []
        if filter_visible:
            # 查询数据集的可见字段
            fields_query = select(DataField).where(
                DataField.dataset_id == dataset_id,
                DataField.visible == '1'
            )
            fields_result = await session.execute(fields_query)
            visible_fields = [field.name for field in fields_result.scalars().all()]
            logger.debug(f"数据集 {dataset_id} 可见字段: {visible_fields}")
        
        # 2. 使用DatasetService获取数据集数据
        result = await DatasetService.get_dataset_data(
            dataset_id=dataset_id, 
            page=page, 
            page_size=page_size, 
            search=search, 
            session=session
        )
        
        if not result.get("success", False):
            raise HTTPException(
                status_code=500,
                detail=result.get("message", "获取数据集数据失败")
            )
        
        # 2.5 如果需要过滤可见字段且有可见字段列表，处理返回的数据
        if filter_visible and visible_fields:
            # 过滤数据中的字段
            filtered_data = []
            for item in result.get("data", []):
                filtered_item = {k: v for k, v in item.items() if k in visible_fields}
                filtered_data.append(filtered_item)
            
            # 更新返回结果
            result["data"] = filtered_data
            
            # 更新显示名称映射，只保留可见字段
            if "column_display_names" in result:
                result["column_display_names"] = {k: v for k, v in result["column_display_names"].items() if k in visible_fields}
            
            # 更新列名列表，只保留可见字段
            if "columns" in result:
                result["columns"] = [col for col in result["columns"] if col in visible_fields]
            
            logger.debug(f"已过滤数据集 {dataset_id} 的数据，只保留可见字段")
            
        # 3. 构建响应，添加额外信息
        # 查询关联的分析维度
        dimensions_query = select(AnalysisDimension).where(AnalysisDimension.dataset_id == dataset_id)
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        
     
        # 添加数据集信息
        result["dataset"] = dataset.to_dict()
        result["analysis_dimensions"] = dimensions

        return result
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取数据集数据失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(
            status_code=500,
            detail=f"获取数据集数据失败: {str(e)}"
        )

   

@router.post("/create_session", response_model=dict)
async def create_session(
    dataset_id: str = Form(None, description="数据集ID，可选"),
    session_title: str = Form(None, description="会话标题，为空时使用默认标题或数据集名称"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    创建一个新的会话，数据集ID为可选参数。
    
    Args:
        dataset_id: 数据集ID，可选。如果提供，会话将关联该数据集
        session_title: 会话标题，可选。如果未提供，将使用数据集名称或默认标题
    """
    try:
        dataset = None
        dataset_name = None
        
        # 1. 如果提供了数据集ID，检查数据集是否存在
        if dataset_id:
            dataset_query = select(DataSet).where(DataSet.id == dataset_id)
            dataset_result = await session.execute(dataset_query)
            dataset = dataset_result.scalar_one_or_none()
            
            if not dataset:
                raise HTTPException(
                    status_code=404,
                    detail="数据集不存在"
                )
            dataset_name = dataset.name
                
        # 2. 创建新会话
        # 确定会话标题
        if not session_title:
            if dataset:
                session_title = dataset.name
            else:
                session_title = "新建会话"
        
        # 创建新会话
        new_session = UserSession(
            user_id=str(current_user.id),
            file_name=session_title,
            session_title=session_title,
            file_url="",
            dataset_ids=dataset_id if dataset_id else "",  # 存储关联的数据集ID，可为空
            upload_time=datetime.now()
        )
        
        # 保存会话
        session.add(new_session)
        await session.commit()
        await session.refresh(new_session)
        
        # 3. 构建响应
        return {
            "success": True,
            "session_id": new_session.id,
            "session_title": new_session.session_title,
            "dataset_id": dataset_id,
            "dataset_name": dataset_name if dataset_name else None
        }
    
    except HTTPException as e:
        # 重新抛出HTTP异常
        raise
    except Exception as e:
        logger.error(f"创建数据集会话失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(
            status_code=500,
            detail=f"创建会话失败: {str(e)}"
        )
  
   

@router.post("/adjust_chart_config", response_model=None)
async def adjust_chart_config(
    message_id: str = Form(..., description="消息ID"),
    data_source_name: str = Form(..., description="数据源名称"),
    user_adjustment: str = Form(..., description="用户对图表的调整意见"),
    user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据用户的自然语言调整意见重新生成图表配置
    
    参数:
        message_id (str): 包含原始图表配置的消息ID
        data_source_name (str): 要调整的数据源名称
        user_adjustment (str): 用户的调整意见，例如"把柱状图改成饼图"、"将颜色改为红色系"等
    
    返回:
        dict: 包含调整后的图表配置
    """
    try:
        # 查询指定ID的消息
        query = select(UserMsg).where(UserMsg.id == message_id)
        result = await session.execute(query)
        message = result.scalar_one_or_none()
        
        if not message:
            raise HTTPException(status_code=404, detail=f"消息不存在: {message_id}")
        
        # 查询会话信息
        session_id = message.session_id
        query = select(UserSession).where(UserSession.id == session_id)
        result = await session.execute(query)
        user_session = result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(status_code=404, detail=f"会话不存在: {session_id}")
        
        # 从消息中获取现有的图表配置
        existing_chart_data = []
        if message.chart_option:
            try:
                existing_chart_data = json.loads(message.chart_option)
            except Exception as e:
                logger.error(f"解析消息中的图表配置失败: {str(e)}")
                raise HTTPException(status_code=400, detail=f"消息中的图表配置格式不正确: {str(e)}")
        
        if not existing_chart_data:
            raise HTTPException(status_code=400, detail="消息中没有现有的图表配置")
        
        # 从消息中提取SQL语句
        all_sql_statements = []
        if message.message_sql:
            try:
                all_sql_statements = json.loads(message.message_sql)
            except Exception as e:
                logger.error(f"解析消息中的SQL失败: {str(e)}")
                all_sql_statements = [message.message_sql]  # 如果解析失败，将整个内容作为一条SQL语句
        
        # 找到特定数据源的现有配置和SQL
        target_chart_config = None
        target_sql = ""
        target_chart_type = ""
        target_config_index = -1
        
        for i, chart_item in enumerate(existing_chart_data):
            if chart_item.get("数据源名称", "") == data_source_name:
                target_chart_config = chart_item
                target_config_index = i
                
                # 查找对应的SQL
                if i < len(all_sql_statements):
                    target_sql = all_sql_statements[i]
                
                # 获取原始图表类型
                target_chart_type = chart_item.get("推荐图表类型", "")
                break
        
        if target_chart_config is None:
            raise HTTPException(status_code=404, detail=f"未找到名为 '{data_source_name}' 的数据源配置")
        
        # 查询源消息
        source_message = None
        if message.message_sourceid:
            source_query = select(UserMsg).where(UserMsg.id == message.message_sourceid)
            source_result = await session.execute(source_query)
            source_message = source_result.scalar_one_or_none()
        
        # 使用源消息的内容，如果没有找到则使用当前消息内容
        original_request = source_message.message if source_message else message.message
        
        # 组合原始请求和调整意见
        combined_request = f"{original_request}。调整要求：{user_adjustment}"
        
        try:
            # 根据目前图表类型，使用相应的方法重新生成配置
            if "g2_config" in target_chart_config and target_chart_config["g2_config"]:
                # 如果有G2配置，重新生成G2配置
                adjusted_g2_config = await run_in_threadpool(
                    advisor.generate_g2_config,
                    target_sql,
                    target_chart_type,
                    combined_request,  # 组合请求
                    None,  # 不传递数据
                    target_chart_config.get("g2_config", "")  # 传递原配置作为参考
                )
                
                # 更新配置
                target_chart_config["g2_config"] = adjusted_g2_config  
            # 更新原配置数组中的项目
            existing_chart_data[target_config_index] = target_chart_config
            
            # 更新消息的图表配置
            message.chart_option = DataHelp.to_json(existing_chart_data)
            session.add(message)
            await session.commit()
            
            # 返回调整后的配置
            return {
                "status": "success",
                "message": "图表配置已根据用户意见调整",
                "message_id": message_id,
                "data_source_name": data_source_name,
                "adjusted_chart_data": target_chart_config
            }
            
        except Exception as e:
            logger.error(f"调整图表配置失败: {str(e)}")
            logger.exception(e)
            raise HTTPException(status_code=500, detail=f"调整图表配置失败: {str(e)}")
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"处理图表调整请求失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"处理图表调整请求失败: {str(e)}")


@router.post("/add_report", response_model=dict)
async def add_report(
    dataset_id: str = Form(..., description="数据集ID"),
    report_title: str = Form(None, description="分析报告标题，可选"),
    description: str = Form(None, description="分析报告描述，可选"),
    filters: str = Form(None, description="数据集过滤条件，JSON格式字符串，可选"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    创建新的分析报告记录，并将数据集的分析维度复制到新报告
    
    参数:
        dataset_id (str): 数据集ID
        report_title (str, optional): 分析报告标题，如果未提供则使用默认值
        description (str, optional): 分析报告描述，如果未提供则使用默认值
        filters (str, optional): 数据集过滤条件，JSON格式字符串，可选
        
    返回:
        dict: 操作结果，包含创建的报告ID和信息
    """
    try:
        # 查询数据集信息
        dataset_query = select(DataSet).where(DataSet.id == dataset_id)
        dataset_result = await session.execute(dataset_query)
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(status_code=404, detail=f"关联的数据集不存在: {dataset_id}")
        
        # 创建分析报告记录
        new_report = AnalysisReport(
            dataset_id=dataset_id,
            report='',
            creator_id=str(user.id),
            report_title=report_title or dataset.name,
            report_dimension="综合分析",
            report_satus="dwc",
            report_description=description or "数据分析报告",
            data_filters=filters,  # 存储过滤条件
            created_at=datetime.now()
        )
        
        session.add(new_report)
        await session.flush()  # 提交以获取报告ID
        await session.refresh(new_report)
        
        # 创建默认的"数据概览"分析维度
        data_overview_dimension = AnalysisDimension(
            dataset_id=dataset_id,
            report_id=new_report.id,
            dimension_name="数据概览",
            dimension_description="数据集的整体概览，包括数据规模、结构特征和主要统计指标",
            analysis_goal="数据概览",
            dimension_type="report",
            creator_id=str(user.id),
            is_locked=0,
            dimension_order=0  # 确保数据概览维度排在最前面
        )
        session.add(data_overview_dimension)
        
        # 查询数据集关联的分析维度
        dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.dataset_id == dataset_id,
            AnalysisDimension.report_id == None  # 只选择没有关联报告的维度（数据集自身的维度）
        )
        dimensions_result = await session.execute(dimensions_query)
        dataset_dimensions = dimensions_result.scalars().all()
        
        # 复制数据集的分析维度作为报告的分析维度
        copied_dimensions_count = 0
        for dimension in dataset_dimensions:
            # 创建分析维度副本
            new_dimension = AnalysisDimension(
                dataset_id=dataset_id,
                report_id=new_report.id,
                dimension_name=dimension.dimension_name,
                dimension_description=dimension.dimension_description,
                analysis_goal=dimension.analysis_goal,
                analysis_method=dimension.analysis_method,
                analysis_sql=dimension.analysis_sql,
                answerable_questions=dimension.answerable_questions,
                chart_suggestions=dimension.chart_suggestions,
                recommended_charts=dimension.recommended_charts,
                json_data=dimension.json_data,
                analysis_result=dimension.analysis_result,
                dimension_type="report",  # 标记为报告维度
                creator_id=str(user.id),  # 更新为当前用户
                is_locked=0,  # 新复制的维度默认不锁定
                dimension_order=copied_dimensions_count + 1  # 设置顺序，确保数据概览维度排在第一位
            )
            
            session.add(new_dimension)
            copied_dimensions_count += 1

        await session.commit()
        return {
            "status": True,
            "message": f"报告创建成功，已添加数据概览维度并复制{copied_dimensions_count}个分析维度",
            "report_id": new_report.id,
            "report": new_report,
            "copied_dimensions_count": copied_dimensions_count
        }
        
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        raise HTTPException(status_code=500, detail=f"创建报告失败: {str(e)}")

@router.post("/generate_analysis_outline", response_model=dict)
async def generate_analysis_outline(
    report_id: str = Form(..., description="报告ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据报告ID生成数据分析大纲并保存到分析维度表
    
    参数:
        report_id (str): 报告ID
        dataset_id (str, optional): 数据集ID，可选。如果不提供，将尝试从现有维度中获取
        
    返回:
        dict: 操作结果，包含生成的分析大纲JSON
    """
    try:
        # 查询报告信息
        report_query = select(AnalysisReport).where(AnalysisReport.id == report_id)
        result = await session.execute(report_query)
        report = result.scalar_one_or_none()
        
        if not report:
            raise HTTPException(status_code=404, detail=f"报告不存在: {report_id}")
        
        # 检查权限：报告创建者可以直接访问
        if report.creator_id != str(user.id):
            raise HTTPException(status_code=403, detail="无权访问此报告")
        
        
        # 查询数据集
        dataset_query = select(DataSet).where(DataSet.id == report.dataset_id)
        dataset_result = await session.execute(dataset_query)
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(status_code=404, detail=f"关联的数据集不存在: {report.dataset_id}")
        
        # 从数据集获取数据结构信息
        if not dataset.dataset_summary:
            raise HTTPException(status_code=400, detail="数据集摘要信息为空")
        
        # 获取该数据集下现有的分析维度
        existing_dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == report.id 
        )
        existing_dimensions_result = await session.execute(existing_dimensions_query)
        existing_dimensions = existing_dimensions_result.scalars().all()
        
        # 将现有维度转换为列表格式
        existing_dimensions_info = []
        for dim in existing_dimensions:
            dim_info = {
                "分析维度(子节标题)": dim.dimension_name,
                "分析说明": dim.dimension_description or ""
            }
            existing_dimensions_info.append(dim_info)
        
        # 生成分析大纲
        try:
            # 使用专用方法生成分析大纲，传递现有维度信息作为上下文
            existing_dimensions_context = json.dumps({"existing_dimensions": existing_dimensions_info}, ensure_ascii=False)
            outline_data = advisor.generate_and_save_analysis_outline(
                dataset.dataset_summary, 
                report.report_description, 
                existing_dimensions_context
            )
        except ValueError as e:
            raise HTTPException(status_code=500, detail=str(e))
        
        # 更新报告的分析大纲字段
        report.analysis_outline = json.dumps(outline_data)
        session.add(report)
        
        # 处理分析大纲中的核心分析章节
        if "核心分析章节" in outline_data:
            for chapter in outline_data["核心分析章节"]:
                chapter_title = chapter.get("章节标题", "")
                # 对每个子节创建一个分析维度记录
                for section in chapter.get("子节", []):
                    section_title = section.get("子节标题", "")
                    
                    # 检查是否已存在类似的分析维度（通过名称判断）
                    similar_dimension_exists = False
                    for existing_dim in existing_dimensions:
                        # 如果找到名称相似度超过70%的维度，则跳过
                        if existing_dim.dimension_name and section_title and (
                            existing_dim.dimension_name.lower() == section_title.lower() or
                            (len(existing_dim.dimension_name) > 3 and existing_dim.dimension_name.lower() in section_title.lower()) or
                            (len(section_title) > 3 and section_title.lower() in existing_dim.dimension_name.lower())
                        ):
                            similar_dimension_exists = True
                            break
                    
                    if similar_dimension_exists:
                        continue
                    
                    # 创建分析维度记录
                    dimension = AnalysisDimension(
                        dataset_id=report.dataset_id,
                        report_id=report_id,
                        dimension_name=section_title,
                        dimension_description=section.get("分析说明", ""),
                        analysis_goal=chapter_title,
                        analysis_method="数据分析",
                        dimension_type="report",
                        dimension_order=1,
                        creator_id=str(user.id),  # 添加当前用户ID作为创建人
                        is_locked=0,
                        status="pending"
                    )
                    
                    session.add(dimension)
                    await session.flush()
        
        # 提交事务
        await session.commit()
        
        return {
            "status": True,
            "message": "分析大纲生成成功",
            "analysis_outline": outline_data,
            "report_id": report_id
        }
        
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        raise HTTPException(status_code=500, detail=f"生成分析大纲失败: {str(e)}")

@router.post("/generate_dimension_fx",response_model=None)
async def generate_dimension_fx(
    dimension_id: str = Form(..., description="分析维度ID"),
    user_opinion: str = Form(..., description="用户意见"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据分析维度ID生成对应的SQL语句
    
    参数:
        dimension_id (str): 分析维度ID
        
    返回:
        dict: 包含生成的SQL语句和其他相关信息
    """
    try:
        # 1. 查询分析维度信息
        query = select(AnalysisDimension).where(AnalysisDimension.id == dimension_id)
        result = await session.execute(query)
        dimension = result.scalar_one_or_none()
        
        if not dimension:
            raise HTTPException(status_code=404, detail=f"分析维度不存在: {dimension_id}")
            
        # 2. 获取报告数据筛选条件（如果维度关联了报告）
        filter_conditions = []
        if dimension.report_id:
            report_query = select(AnalysisReport).where(AnalysisReport.id == dimension.report_id)
            report_result = await session.execute(report_query)
            report = report_result.scalar_one_or_none()
            
            if report and report.data_filters:
                try:
                    # 解析 data_filters 字段
                    filters_data = json.loads(report.data_filters)
                    
                    # 提取 description 属性
                    if isinstance(filters_data, list):
                        for filter_item in filters_data:
                            if isinstance(filter_item, dict) and 'description' in filter_item:
                                filter_conditions.append(filter_item['description'])
                except json.JSONDecodeError:
                    logger.error(f"无法解析报告 {dimension.report_id} 的数据筛选条件")
                except Exception as e:
                    logger.error(f"处理报告筛选条件时发生错误: {str(e)}")
        
        # 查询数据集
        dataset_query = select(DataSet).where(DataSet.id == dimension.dataset_id)
        dataset_result = await session.execute(dataset_query)
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(status_code=404, detail=f"数据集不存在: {dimension.dataset_id}")
        
        if dataset and dataset.dataset_summary:
            # 使用数据集的dataset_summary字段
            table_info = json.loads(dataset.dataset_summary)
        
        # 如果没有获取到表结构信息，则无法生成SQL
        if not table_info:
            raise HTTPException(status_code=400, detail="无法获取关联数据集的结构信息")
        
        # 检查是否为"数据概览"分析维度
        if dimension.dimension_name == "数据概览":
            # 对于数据概览，使用与generate_report接口类似的方法获取摘要报告
            report_data = await DataHelp.generate_dataset_summary(dimension.dataset_id, session)
            
            if not report_data:
                raise HTTPException(status_code=400, detail="生成数据概览失败，数据集摘要为空")
            
            # 将数据摘要转换为JSON字符串
            report_data_json = DataHelp.to_json(report_data)
            
            # 使用generate_report方法生成报告，但需要捕获流式输出并组合
            report_content = ""
            
            # 使用generate_report生成器，捕获所有输出
            for chunk in advisor.generate_report("summary", report_data_json, "数据概览报告"):
                report_content += chunk
            
            # 更新分析维度的结果
            dimension.analysis_result = report_content
            # 可以清空或设置默认值
            dimension.analysis_sql = ""
            dimension.answerable_questions = user_opinion
            dimension.status="completed"
            # 保存更新
            session.add(dimension)
            await session.commit()
            
            return {
                "status": True,
                "dimension": dimension
            }
        
        # 4. 构建查询文本，结合dimension_name、dimension_description和answerable_questions
        query_text = f"{dimension.dimension_name}"
        if dimension.dimension_description:
            query_text += f": {dimension.dimension_description}"
        if dimension.answerable_questions:
            query_text += f" 用户意见: {dimension.answerable_questions}"
            
        # 添加数据筛选条件到查询文本
        if filter_conditions:
            filter_text = "; ".join(filter_conditions)
            query_text += f" 数据筛选条件: {filter_text}"
        
        # 历史消息可以为空，这里传入一个空字符串
        history_messages = ""
        
        # 5. 调用advisor.generate_sql_plan生成SQL
        responseai = await run_in_threadpool(
            advisor.generate_sql_plan,
            table_info, 
            query_text,
            history_messages
        )
        
        if not responseai or len(responseai) == 0:
            raise HTTPException(status_code=400, detail="生成SQL失败，大模型返回结果为空")
            
        # 6. 从响应中提取SQL和其他信息
        # 获取数据源信息
        data_sources = responseai[0].get("数据源信息", [])
        sql_statements = []
        chart_types = []
        sql_explanations = []
        data_source_names = []
        
        for source in data_sources:
            sql = source.get("SQL", "")
            if sql:
                sql_statements.append(sql)
                chart_types.append(source.get("推荐图表类型", ""))
                sql_explanations.append(source.get("SQL说明", ""))
                data_source_names.append(source.get("数据源名称", "未命名数据源"))
        
        # 7. 更新分析维度的SQL字段
        all_sql = ";\n\n".join(sql_statements)
        dimension.analysis_sql = all_sql
        
        # 执行 SQL 获取数据
        if sql_statements:
            result = await DatasetService.execute_sql_from_source(dataset.data_source_id, sql_statements[0], session)
            message_data = result.get("data", [])
            dimension.json_data = DataHelp.to_json(message_data)
            
            # 生成分析报告
            responseai = await run_in_threadpool(
                advisor.generate_interpretation_report,
                table_info,
                message_data,
                dimension.dict()
            )
            
            dimension.analysis_result = responseai
        
        dimension.answerable_questions = user_opinion
        dimension.status="completed"
        # 8. 保存更新
        session.add(dimension)
        await session.commit()
        
        return {
            "status": True,
            "dimension": dimension
        }
     
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"生成分析维度SQL失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"生成分析维度SQL失败: {str(e)}")

@router.post("/get_my_reports", response_model=dict)
async def get_my_reports(
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据当前用户获取该用户创建的所有报告
    
    参数:
        user (User): 当前用户
        
    返回:
        List[dict]: 报告列表
    """
    try:
       
        # 查询该会话的所有报告，按创建时间倒序排序
        reports_query = select(AnalysisReport).where(AnalysisReport.creator_id == str(user.id)).order_by(AnalysisReport.created_at.desc())
        reports_result = await session.execute(reports_query)
        reports = reports_result.scalars().all()
        
        # 格式化返回数据
        reports_data={
            "status":"success",
            "data":reports
        }
        return reports_data
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取会话报告失败: {str(e)}")

@router.post("/get_report_details",response_model=dict)
async def get_report_details(
    report_id: str = Form(..., description="报告ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据报告ID获取报告详细信息和相关章节
    
    参数:
        report_id (str): 报告ID
        
    返回:
        dict: 包含报告信息和章节列表的字典
    """
    try:
        # 查询报告信息
        report_query = select(AnalysisReport).where(AnalysisReport.id == report_id)
        result = await session.execute(report_query)
        report = result.scalar_one_or_none()
        
        if not report:
            raise HTTPException(status_code=404, detail=f"报告不存在: {report_id}")
        
        # 检查权限：报告创建者可以直接访问
        if report.creator_id == str(user.id):
            has_permission = True
        else:
            # 如果不是创建者，检查会话是否属于当前用户
            user_session_query = select(UserSession).where(
                UserSession.id == report.session_id,
                UserSession.user_id == str(user.id)
            )
            session_result = await session.execute(user_session_query)
            user_session = session_result.scalar_one_or_none()
            has_permission = user_session is not None
        
        if not has_permission:
            raise HTTPException(status_code=403, detail="无权访问此报告")
        
        # 查询该报告关联的所有章节（分析维度）
        dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == report_id
        ).order_by(AnalysisDimension.dimension_order,AnalysisDimension.created_at)
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        
        # 格式化返回数据
        report_data = {
            "status":"success",
            "report":report,
            "dimensions":dimensions
        }
        return report_data
    
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"获取报告详情失败: {str(e)}")

@router.post("/delete_report", response_model=dict)
async def delete_report(
    report_id: str = Form(..., description="报告ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据报告ID删除报告及其关联的章节信息
    
    参数:
        report_id (str): 报告ID
        
    返回:
        dict: 操作结果
    """
    try:
        # 查询报告信息
        report_query = select(AnalysisReport).where(AnalysisReport.id == report_id)
        result = await session.execute(report_query)
        report = result.scalar_one_or_none()
        
        if not report:
            raise HTTPException(status_code=404, detail=f"报告不存在: {report_id}")
        
        # 检查权限：报告创建者可以直接删除
        if report.creator_id == str(user.id):
            has_permission = True
        else:
            # 如果不是创建者，检查会话是否属于当前用户
            user_session_query = select(UserSession).where(
                UserSession.id == report.session_id,
                UserSession.user_id == str(user.id)
            )
            session_result = await session.execute(user_session_query)
            user_session = session_result.scalar_one_or_none()
            has_permission = user_session is not None
        
        if not has_permission:
            raise HTTPException(status_code=403, detail="无权删除此报告")
        
        # 1. 先删除报告关联的所有章节（分析维度）
        dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == report_id
        )
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        
        # 记录删除的章节数量
        deleted_dimensions_count = len(dimensions)
        
        # 删除所有关联的章节
        for dimension in dimensions:
            await session.delete(dimension)
        
        # 2. 删除报告本身
        await session.delete(report)
        
        # 提交事务
        await session.commit()
        
        return {
            "status": "success",
            "message": f"报告已成功删除，同时删除了 {deleted_dimensions_count} 个关联章节",
            "report_id": report_id
        }
    
    except HTTPException:
        # 发生HTTP异常时，回滚事务并继续抛出异常
        await session.rollback()
        raise
    except Exception as e:
        # 发生其他异常时，回滚事务并记录日志
        await session.rollback()
        logger.error(f"删除报告失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"删除报告失败: {str(e)}")

@router.post("/delete_analysis_dimension", response_model=dict)
async def delete_analysis_dimension(
    dimension_id: str = Form(..., description="分析维度ID，多个ID以逗号分隔"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    删除指定的分析维度，支持批量删除
    
    参数:
        dimension_id (str): 分析维度ID，多个ID以逗号分隔
        
    返回:
        dict: 操作结果
    """
    try:
        # 拆分维度ID，支持多个ID以逗号分隔
        dimension_ids = [id.strip() for id in dimension_id.split(',') if id.strip()]
        
        if not dimension_ids:
            raise HTTPException(status_code=400, detail="未提供有效的分析维度ID")
        
        deleted_count = 0
        failed_ids = []
        deleted_names = []
        
        # 逐个处理每个维度ID
        for dim_id in dimension_ids:
            try:
                # 查询指定ID的分析维度
                query = select(AnalysisDimension).where(AnalysisDimension.id == dim_id)
                result = await session.execute(query)
                dimension = result.scalar_one_or_none()
                
                if not dimension:
                    failed_ids.append(f"{dim_id}(不存在)")
                    continue
                
                # 如果当前用户是创建人，直接允许删除
                if dimension.creator_id == str(user.id):
                    dimension_name = dimension.dimension_name
                    await session.delete(dimension)
                    deleted_count += 1
                    deleted_names.append(dimension_name)
                else:
                    failed_ids.append(f"{dim_id}(无权限)")
            except Exception as e:
                logger.error(f"删除分析维度 {dim_id} 时出错: {str(e)}")
                failed_ids.append(f"{dim_id}(错误)")
        
        # 提交事务，一次性保存所有删除操作
        if deleted_count > 0:
            await session.commit()
        
        # 构建响应消息
        message = f"成功删除 {deleted_count} 个分析维度"
        if deleted_names:
            message += f": {', '.join(deleted_names[:3])}"
            if len(deleted_names) > 3:
                message += f" 等{len(deleted_names)}个维度"
                
        if failed_ids:
            message += f"，{len(failed_ids)} 个维度删除失败: {', '.join(failed_ids[:5])}"
            if len(failed_ids) > 5:
                message += f" 等{len(failed_ids)}个维度"
        
        return {
            "status": "success" if deleted_count > 0 else "partial" if failed_ids else "failed",
            "message": message,
            "deleted_count": deleted_count,
            "failed_count": len(failed_ids),
            "total_count": len(dimension_ids)
        }
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        await session.rollback()
        logger.error(f"批量删除分析维度失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"删除分析维度失败: {str(e)}")

@router.post("/generate_report_html", response_model=dict)
async def generate_report_html(
    report_id: str = Form(..., description="报告ID"),
    user_requirements: str = Form("", description="用户对报告的生成要求，可选"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据报告ID获取已经生成分析内容的分析维度，将分析内容拼接成一个Markdown格式的文档，
    然后使用生成HTML模板并填充内容的方式生成HTML报告，并保存到报表表的report字段中
    
    参数:
        report_id (str): 报告ID
        user_requirements (str, optional): 用户对报告生成的特定要求，如样式、图表类型等
        
    返回:
        dict: 操作结果，包含成功状态和消息
    """
    try:
        # 1. 查询报告信息
        report_query = select(AnalysisReport).where(AnalysisReport.id == report_id)
        result = await session.execute(report_query)
        report = result.scalar_one_or_none()
        
        if not report:
            raise HTTPException(status_code=404, detail=f"报告不存在: {report_id}")
        
        # 2. 检查权限：报告创建者可以直接访问
        if report.creator_id == str(user.id):
            has_permission = True
        if not has_permission:
            raise HTTPException(status_code=403, detail="无权访问此报告")
        
        # 3. 查询该报告关联的所有章节（分析维度），并且已经生成了分析结果的维度
        dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == report_id,
            AnalysisDimension.analysis_result.isnot(None),
            AnalysisDimension.analysis_result != ""
        ).order_by(AnalysisDimension.created_at)
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        
        if not dimensions:
            raise HTTPException(status_code=400, detail="报告没有已生成内容的分析维度")
        
        # 5. 拼接Markdown格式的报告内容
        md_content = f"# {report.report_title or '数据分析报告'}\n\n"
        
        # 添加报告描述
        if report.report_description:
            md_content += f"## 报告概述\n\n{report.report_description}\n\n"
        
        # 添加创建时间
        if report.created_at:
            md_content += f"**生成时间**: {report.created_at.strftime('%Y-%m-%d %H:%M:%S')}\n\n"
        
        # 创建章节目录
        md_content += "## 目录\n\n"
        
        # 直接按顺序列出所有维度作为目录
        toc_content = ""
        for i, dimension in enumerate(dimensions):
            toc_item = f"{i+1}. [{dimension.dimension_name}](#chapter-{i+1})\n"
            md_content += toc_item
            toc_content += toc_item
        
        md_content += "\n"
        
        # 按顺序添加所有维度内容
        content_md = ""
        for i, dimension in enumerate(dimensions):
            content_md += f"<a id='chapter-{i+1}'></a>\n\n"
            content_md += f"# {i+1}. {dimension.dimension_name}\n\n"
            
            # 如果有维度描述，添加描述
            if dimension.dimension_description:
                content_md += f"*{dimension.dimension_description}*\n\n"
                
            # 添加分析结果内容
            if dimension.analysis_result:
                content_md += dimension.analysis_result + "\n\n"
        
        # 6. 使用优化的方法生成HTML
        try:
            # 添加报告相关上下文信息
            additional_context = f"""
            报告标题: {report.report_title or '数据分析报告'}
            报告描述: {report.report_description or ''}
            报告维度: {report.report_dimension or '综合分析'}
            生成时间: {report.created_at.strftime('%Y-%m-%d %H:%M:%S') if report.created_at else ''}
            """
            
            # 步骤1: 使用AI生成HTML模板
            html_template = await run_in_threadpool(
                advisor.generate_html_template,
                user_requirements,
                additional_context
            )
            
            # 步骤2: 将Markdown内容填充到模板中
            report_data = {
                'title': report.report_title or '数据分析报告',
                'description': report.report_description or '',
                'content': content_md,
                'toc': toc_content,
                'created_at': report.created_at.strftime('%Y-%m-%d %H:%M:%S') if report.created_at else ''
            }
            
            html_content = await run_in_threadpool(
                advisor.fill_html_template,
                html_template,
                report_data
            )
            
            # 7. 更新报告的HTML内容
            report.report = html_content
            report.report_satus = "done"  # 更新报告状态为完成
            
            # 8. 保存到数据库
            session.add(report)
            await session.commit()
            
            return {
                "status": "success",
                "message": "报告HTML生成成功",
                "report_id": report_id
            }
            
        except Exception as e:
            logger.error(f"生成HTML报告失败: {str(e)}")
            raise HTTPException(status_code=500, detail=f"生成HTML报告失败: {str(e)}")
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"生成报告HTML失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"生成报告HTML失败: {str(e)}")

@router.post("/delete_my_dataset", response_model=dict)
async def delete_my_dataset(
    dataset_id: str = Form(..., description="数据集ID"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    删除当前用户创建的数据集
    
    只能删除自己创建的数据集，不能删除通过角色权限获得的数据集
    
    Args:
        dataset_id: 数据集ID
    """
    try:
        # 查询数据集
        result = await session.execute(
            select(DataSet).filter(DataSet.id == dataset_id)
        )
        dataset = result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(
                status_code=404,
                detail=f"数据集不存在: {dataset_id}"
            )
        
        # 验证是否是用户自己创建的数据集
        if dataset.user_id != str(current_user.id):
            raise HTTPException(
                status_code=403,
                detail="您只能删除自己创建的数据集"
            )
            
        # 查询分析维度记录
        dimensions_query = select(AnalysisDimension).filter(
            AnalysisDimension.dataset_id == dataset_id
        )
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        
        # 删除相关的分析维度记录
        for dimension in dimensions:
            await session.delete(dimension)
        await session.delete(dataset)
        await session.commit()
        
        return {
            "status": True,
            "message": "数据集删除成功"
        }
    
    except HTTPException:
        raise
    
    except Exception as e:
        await session.rollback()
        raise HTTPException(
            status_code=500,
            detail=f"删除数据集失败: {str(e)}"
        )

@router.post("/dataset-info", response_model=dict)
async def get_dataset_info(
    dataset_id: str = Form(..., description="数据集ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据数据集ID获取数据集的基本信息和字段信息
    
    参数:
        dataset_id (str): 数据集ID
        
    返回:
        dict: 包含数据集基本信息和字段信息的字典
    """
    try:
        # 1. 查询数据集信息
        dataset_query = select(DataSet).where(DataSet.id == dataset_id)
        dataset_result = await session.execute(dataset_query)
        dataset = dataset_result.scalar_one_or_none()
        
        if not dataset:
            raise HTTPException(status_code=404, detail=f"数据集不存在: {dataset_id}")
        
      
        # 3. 获取字段信息
        fields_query = select(DataField).where(
            DataField.dataset_id == dataset_id,
            DataField.visible == "1"
        ).order_by(DataField.ordinal_position)
        fields_result = await session.execute(fields_query)
        fields = fields_result.scalars().all()
        
        # 4. 获取分析维度信息
        dimensions_query = select(AnalysisDimension).where(
            AnalysisDimension.dataset_id == dataset_id,
            AnalysisDimension.report_id == None  # 仅获取数据集自身的维度，不包括报告维度
        )
        dimensions_result = await session.execute(dimensions_query)
        dimensions = dimensions_result.scalars().all()
        return {
            "status": True,
            "message": "获取数据集信息成功",
            "dataset": dataset,
            "fields": fields,
            "analysis_dimensions": dimensions
        }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取数据集信息失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"获取数据集信息失败: {str(e)}")

@router.post("/update_analysis_result", response_model=dict)
async def update_analysis_result(
    dimension_id: str = Form(..., description="分析维度ID，多个ID以逗号分隔"),
    analysis_result: str = Form(..., description="分析结果内容"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    更新分析维度的analysis_result字段，支持批量更新
    
    参数:
        dimension_id (str): 分析维度ID，多个ID以逗号分隔
        analysis_result (str): 分析结果内容
        
    返回:
        dict: 操作结果
    """
    try:
        # 拆分维度ID，支持多个ID以逗号分隔
        dimension_ids = [id.strip() for id in dimension_id.split(',') if id.strip()]
        
        if not dimension_ids:
            raise HTTPException(status_code=400, detail="未提供有效的分析维度ID")
        
        updated_count = 0
        failed_ids = []
        updated_names = []
        
        # 逐个处理每个维度ID
        for dim_id in dimension_ids:
            try:
                # 查询指定ID的分析维度
                query = select(AnalysisDimension).where(AnalysisDimension.id == dim_id)
                result = await session.execute(query)
                dimension = result.scalar_one_or_none()
                
                if not dimension:
                    failed_ids.append(f"{dim_id}(不存在)")
                    continue
                
                # 检查权限：当前用户是否为创建者
                if dimension.creator_id and dimension.creator_id != str(user.id):
                    failed_ids.append(f"{dim_id}(无权限)")
                    continue
                
                # 更新分析结果
                dimension.analysis_result = analysis_result
                if dimension.analysis_result == "":
                    dimension.status = "pending"
                else:
                    dimension.status = "completed"
                
                updated_count += 1
                updated_names.append(dimension.dimension_name)
                
            except Exception as e:
                logger.error(f"更新分析维度 {dim_id} 时出错: {str(e)}")
                failed_ids.append(f"{dim_id}(错误)")
        
        # 提交事务，一次性保存所有更新操作
        if updated_count > 0:
            await session.commit()
        
        return {
            "status": True,
            "message": "分析结果更新成功"
        }
        
    except HTTPException:
        await session.rollback()
        raise
    except Exception as e:
        logger.error(f"批量更新分析结果失败: {str(e)}")
        await session.rollback()
        raise HTTPException(status_code=500, detail=f"更新分析结果失败: {str(e)}")

@router.post("/update_report_info", response_model=dict)
async def update_report_info(
    report_id: str = Form(..., description="报告ID"),
    report_title: str = Form(..., description="报告标题"),
    report_description: str = Form(..., description="报告描述"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    更新分析报告的标题和描述
    
    参数:
        report_id (str): 报告ID
        report_title (str): 新的报告标题
        report_description (str): 新的报告描述
        
    返回:
        dict: 操作结果
    """
    try:
        # 查询指定ID的报告
        query = select(AnalysisReport).where(AnalysisReport.id == report_id)
        result = await session.execute(query)
        report = result.scalar_one_or_none()
        
        if not report:
            raise HTTPException(status_code=404, detail=f"报告不存在: {report_id}")
        
        # 检查权限：当前用户是否为创建者
        if report.creator_id and report.creator_id != str(user.id):
            # 检查用户是否有权限修改
            raise HTTPException(status_code=403, detail="无权修改此报告")
        
        # 更新报告标题和描述
        report.report_title = report_title
        report.report_description = report_description
        await session.commit()
        
        return {
            "status": True,
            "message": "报告信息更新成功",
            "report_id": report_id
        }
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"更新报告信息失败: {str(e)}")
        await session.rollback()
        raise HTTPException(status_code=500, detail=f"更新报告信息失败: {str(e)}")

@router.post("/task_breakdown", response_model=dict)
async def task_breakdown(
    user_requirement: str = Form(..., description="用户需求文本"),
    message_id: str = Form(..., description="消息ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据用户需求拆解任务，返回任务拆解结果。
    支持上下文感知，包括会话历史、用户偏好、历史任务等信息。
    
    参数:
        user_requirement: 用户需求文本
        message_id: 消息ID，用于关联任务
    返回:
        dict: 任务拆解结果
    """
    try:
        # 首先查询消息所属的会话
        msg_query = select(UserMsg).where(UserMsg.id == message_id)
        msg_result = await session.execute(msg_query)
        user_msg = msg_result.scalar_one_or_none()
        
        if not user_msg:
            raise HTTPException(status_code=404, detail="消息不存在")
        
        # 获取消息关联的会话
        session_id = user_msg.session_id
        session_query = select(UserSession).where(UserSession.id == session_id)
        session_result = await session.execute(session_query)
        user_session = session_result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(status_code=404, detail="会话不存在")
        
        # === 构建上下文感知信息 ===
        
        # 1. 获取会话历史（最近10条消息）
        conversation_history = []
        history_query = select(UserMsg).where(
            UserMsg.session_id == session_id,
            UserMsg.timestamp <= user_msg.timestamp  # 只获取当前消息之前的历史
        ).order_by(UserMsg.timestamp.desc()).limit(10)
        history_result = await session.execute(history_query)
        history_messages = history_result.scalars().all()
        
        for msg in reversed(history_messages):  # 按时间正序排列
            conversation_history.append({
                'content': msg.message,
                'role': 'user' if msg.message_type == 'user' else 'assistant',
                'timestamp': msg.timestamp.isoformat() if msg.timestamp else None,
                'intent': msg.message_intent
            })
        
        # 2. 获取用户偏好（可以从用户设置或历史行为中推断）
        user_preferences = {
            'detail_level': 'balanced',  # 默认平衡模式
            'chart_preference': [],      # 图表偏好
            'report_style': 'standard'   # 报告风格
        }
        
        # 从历史消息中分析用户偏好
        chart_keywords = []
        analysis_depth_indicators = []
        
        for msg in history_messages[-5:]:  # 分析最近5条消息
            if msg.message:
                content = msg.message.lower()
                # 分析图表偏好
                if '饼图' in content:
                    chart_keywords.append('饼图')
                if '柱状图' in content or '柱图' in content:
                    chart_keywords.append('柱状图')
                if '折线图' in content or '线图' in content:
                    chart_keywords.append('折线图')
                if '散点图' in content:
                    chart_keywords.append('散点图')
                if '热力图' in content:
                    chart_keywords.append('热力图')
                
                # 分析详细程度偏好
                if any(word in content for word in ['简单', '简洁', '快速', '概览']):
                    analysis_depth_indicators.append('simple')
                elif any(word in content for word in ['详细', '深入', '全面', '完整']):
                    analysis_depth_indicators.append('detailed')
        
        # 更新用户偏好
        if chart_keywords:
            user_preferences['chart_preference'] = list(set(chart_keywords))
        
        if analysis_depth_indicators:
            # 根据最近的偏好指标确定详细程度
            recent_preference = analysis_depth_indicators[-1]
            user_preferences['detail_level'] = recent_preference
        
        # 3. 获取历史任务（从分析维度表中获取）
        previous_tasks = []
        tasks_query = select(AnalysisDimension).where(
            AnalysisDimension.creator_id == str(user.id),
            AnalysisDimension.created_at <= user_msg.timestamp if user_msg.timestamp else datetime.now()
        ).order_by(AnalysisDimension.created_at.desc()).limit(10)
        tasks_result = await session.execute(tasks_query)
        historical_tasks = tasks_result.scalars().all()
        
        for task in historical_tasks:
            task_info = {
                '任务类型': task.analysis_method or '数据查询',
                '任务标题': task.dimension_name,
                'status': task.status or 'unknown',
                '数据集': task.dataset_id,
                '创建时间': task.created_at.isoformat() if task.created_at else None
            }
            previous_tasks.append(task_info)
        
        # 4. 获取会话级别上下文
        session_context = {
            'session_id': session_id,
            'session_title': user_session.session_title,
            'current_focus': '数据分析',  # 可以根据会话内容动态确定
            'analysis_depth': user_preferences['detail_level']
        }
        
        # 5. 构建领域知识上下文（可以根据数据集类型确定）
        domain_knowledge = {
            'domain': '数据分析',
            'expertise_level': 'intermediate'  # 可以根据用户历史行为确定
        }
        
        # 组装完整的上下文信息
        conversation_context = {
            'conversation_history': conversation_history,
            'user_preferences': user_preferences,
            'previous_tasks': previous_tasks,
            'session_context': session_context,
            'domain_knowledge': domain_knowledge
        }
        
        logger.info(f"构建上下文信息完成 - 历史消息: {len(conversation_history)}, 历史任务: {len(previous_tasks)}, 用户偏好: {user_preferences}")
        
        # === 获取可用数据集信息 ===
        available_datasets = []
        
        # 检查会话是否关联了数据集
        if user_session.dataset_ids:
            # 获取会话关联的数据集
            dataset_id_list = user_session.get_dataset_ids()
            
            # 只处理关联的数据集
            for dataset_id in dataset_id_list:
                dataset_query = select(DataSet).where(DataSet.id == dataset_id)
                dataset_result = await session.execute(dataset_query)
                dataset = dataset_result.scalar_one_or_none()
                
                if dataset:
                    # 读取数据集的数据摘要信息
                    data_summary = {}
                    if dataset.dataset_summary:
                        try:
                            data_summary = json.loads(dataset.dataset_summary)
                        except:
                            data_summary = {}
                    
                    dataset_info = {
                        "数据集ID": dataset.id,
                        "数据集名称": dataset.name,
                        "数据集描述": dataset.description,
                        "列信息": data_summary.get("列信息", [])
                    }
                    available_datasets.append(dataset_info)
            
        else:
            # 如果会话没有关联数据集，则获取用户有权限访问的所有数据集
            user_role_codes = await get_user_role_codes(str(user.id), session)
            
            # 查询用户有权限的数据集
            dataset_query = select(DataSet)
            
            # 如果用户不是管理员，需要添加权限过滤
            if "admin" not in user_role_codes:
                # 获取用户的角色ID
                role_query = select(UserRole.role_id).where(UserRole.user_id == str(user.id))
                role_result = await session.execute(role_query)
                role_ids = [row[0] for row in role_result.all()]
                
                # 构建权限过滤条件：数据集是用户创建的或用户通过角色有权限访问
                permission_filter = or_(
                    DataSet.user_id == str(user.id),  # 用户创建的数据集
                    DataSet.id.in_(
                        select(RoleDatasetAccess.dataset_id)
                        .where(RoleDatasetAccess.role_id.in_(role_ids))
                    )  # 用户通过角色有权限访问的数据集
                )
                dataset_query = dataset_query.where(permission_filter)
            
            # 执行查询
            datasets_result = await session.execute(dataset_query)
            datasets = datasets_result.scalars().all()
            
            # 将数据集信息转换为JSON格式
            for dataset in datasets:
                # 读取数据集的数据摘要信息
                data_summary = {}
                if dataset.dataset_summary:
                    try:
                        data_summary = json.loads(dataset.dataset_summary)
                    except:
                        data_summary = {}
                
                dataset_info = {
                    "数据集ID": dataset.id,
                    "数据集名称": dataset.name,
                    "数据集描述": dataset.description,
                    "列信息": data_summary.get("列信息", [])
                }
                available_datasets.append(dataset_info)
                    
        if not available_datasets:
            logger.warning("未找到可用的数据集")
        
        # === 使用上下文感知的任务拆解 ===
        # 直接调用AI生成任务拆解结果，传入上下文信息
        task_breakdown_result = await run_in_threadpool(
            advisor.generate_task_breakdown,
            user_requirement,
            available_datasets,
            conversation_context  # 传递上下文信息给AI
        )
        
        # 保存任务到分析维度表
        saved_tasks_count = 0
        failed_tasks_count = 0
        
        for task in task_breakdown_result:
            # 从任务中获取相关信息
            task_group = task.get("任务组号", "0")
            task_type = task.get("任务类型", "未知")
            task_title = task.get("任务标题", "未命名任务")
            task_requirements = task.get("任务要求", {})
            context_relation = task.get("上下文关联", "")  # 新增的上下文关联信息
            
            # 检查是否为失败任务
            is_failed_task = task_requirements.get("执行状态") == "failed"
            
            # 获取关联的数据集ID（如果有）
            related_dataset_id = "0"
            if user_session.dataset_ids and user_session.get_dataset_ids():
                related_dataset_id = user_session.get_dataset_ids()[0]
            
            # 如果会话没有关联数据集或数据集ID为0，尝试从任务要求中查找
            if related_dataset_id == "0" and task_requirements and not is_failed_task:
                dataset_name = task_requirements.get("数据集", "")
                if dataset_name:
                    # 根据数据集名称查找数据集ID
                    dataset_query = select(DataSet).where(
                        DataSet.name == dataset_name,
                        DataSet.user_id == str(user.id)  # 只查找当前用户的数据集
                    )
                    dataset_result = await session.execute(dataset_query)
                    found_dataset = dataset_result.scalar_one_or_none()
                    
                    if found_dataset:
                        related_dataset_id = found_dataset.id
            
            # 将任务要求转换为JSON字符串
            task_requirements_str = json.dumps(task_requirements, ensure_ascii=False)
            
            # 根据任务状态设置不同的初始状态和结果
            if is_failed_task:
                # 失败任务直接设置为失败状态
                initial_status = "failed"
                analysis_result = task_requirements.get("失败原因", "任务拆解阶段标记为失败")
                failed_tasks_count += 1
                logger.info(f"创建失败任务: {task_title} - {analysis_result}")
            else:
                # 正常任务设置为待处理状态
                initial_status = "pending"
                analysis_result = None
            
            # 创建新的分析维度记录
            new_dimension_id = str(uuid.uuid4())
            stmt = insert(AnalysisDimension).values(
                id=new_dimension_id,
                report_id=message_id,
                dataset_id=related_dataset_id,  # 使用关联的数据集ID
                dimension_name=task_title,  # 任务标题存入dimension_name
                dimension_description=task_requirements_str,  # 任务要求存入dimension_description
                analysis_method=task_type,  # 任务类型存入analysis_method
                dimension_order=task_group,  # 任务组号存入dimension_order
                dimension_type="task",
                answerable_questions=context_relation,  # 上下文关联信息存入answerable_questions字段
                analysis_result=analysis_result,  # 失败任务直接设置失败原因
                creator_id=str(user.id),
                created_at=datetime.now(),
                is_locked=0,
                status=initial_status  # 根据任务类型设置初始状态
            )
            
            await session.execute(stmt)
            saved_tasks_count += 1
        
        # 提交事务
        await session.commit()
        
        # 只对非失败任务执行TaskExecutor
        taskinfo = {"message": "任务创建完成"}
        if saved_tasks_count > failed_tasks_count:
            # 创建任务执行器实例
            task_executor = TaskExecutor()
            
            # 执行任务（TaskExecutor会自动跳过已失败的任务）
            taskinfo = await task_executor.execute_tasks_by_message(message_id, session)
            
            # 强制提交任务执行的更改
            await session.commit()
        
        # 收集数据查询任务的结果（包括完成和失败的任务）
        all_query_results = []
        
        # 查询已执行完成或失败的数据查询任务
        completed_tasks_query = select(AnalysisDimension).where(
            AnalysisDimension.report_id == message_id,
            AnalysisDimension.dimension_type == "task",
            AnalysisDimension.analysis_method == "数据查询",
            AnalysisDimension.status == "completed"
        )
        completed_tasks_result = await session.execute(completed_tasks_query)
        completed_tasks = completed_tasks_result.scalars().all()
        
        # 处理每个完成的数据查询任务
        for task in completed_tasks:
            try:
                # 解析任务要求以获取数据说明
                task_requirements = {}
                datadec = ""
                task_requirements = json.loads(task.dimension_description)
                datadec = task_requirements.get("说明", "")
               
                
                # 解析json_data
                json_data = []
                if task.json_data:
                    try:
                        json_data = json.loads(task.json_data)
                    except json.JSONDecodeError:
                        logger.warning(f"任务 {task.id} 的json_data不是有效的JSON格式")
                        json_data = []
                
                # 构建结果对象
                result_item = {
                    "data": json_data,
                    "datadec": datadec
                }
                
                all_query_results.append(result_item)
                
            except Exception as e:
                continue
        


        
        new_msg = UserMsg(
            user_id=str(user.id),
            session_id=session_id,
            message='系统回复',
            message_type='system',
            message_sourceid=message_id,
            timestamp=datetime.now()
        )
        session.add(new_msg)
        await session.commit()
        
        return {
            "status": True,
            "message": "任务拆解和执行完成",
            "taskinfo": taskinfo,
            "hfmsgid": new_msg.id,
            "tasks_count": len(task_breakdown_result),
            "saved_tasks_count": saved_tasks_count,
            "datasets_count": len(available_datasets),
            "all_query_results": all_query_results,  # 添加数据查询结果
            "context_info": {
                "conversation_history_count": len(conversation_history),
                "previous_tasks_count": len(previous_tasks),
                "user_preferences": user_preferences,
                "session_context": session_context
            }
        }
            
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"任务启动失败: {str(e)}")

@router.post("/execute_tasks", response_model=dict)
async def execute_tasks(
    message_id: str = Form(..., description="消息ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据消息ID执行相关的任务集
    
    参数:
        message_id (str): 消息ID，用于查找相关任务
        
    返回:
        dict: 任务执行结果
    """
    try:
        # 创建任务执行器实例
        task_executor = TaskExecutor()
        
        # 执行任务
        result = await task_executor.execute_tasks_by_message(message_id, session)
        
        return result
        
    except Exception as e:
        logger.error(f"执行任务失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"执行任务失败: {str(e)}")

@router.post("/get_tasks_status", response_model=dict)
async def get_tasks_status(
    message_id: str = Form(..., description="消息ID"),
    user: User = Depends(fastapi_users.current_user(active=True)),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据消息ID获取任务执行状况
    
    参数:
        message_id (str): 消息ID，用于查找相关任务
        
    返回:
        dict: 任务执行状况
    """
    try:
        # 创建任务执行器实例
        task_executor = TaskExecutor()
        
        # 获取任务状况
        result = await task_executor.get_tasks_status_by_message(message_id, session)
        
        return result
        
    except Exception as e:
        logger.error(f"获取任务状况失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(status_code=500, detail=f"获取任务状况失败: {str(e)}")

@router.post("/get_temp_session", response_model=dict)
async def get_temp_session(
    dataset_id: str = Form(None, description="数据集ID，可选，用于创建会话时关联数据集"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    获取或创建用户临时会话
    
    Args:
        dataset_id: 数据集ID，可选，用于创建会话时关联数据集
    
    逻辑:
        - 查找当前用户是否有临时会话（session_remark="临时会话"）
        - 如果有临时会话就返回该会话
        - 如果没有则创建一个新的临时会话，并关联指定的数据集（如果提供了dataset_id）
    """
    try:
        # 查找现有的临时会话
        query = select(UserSession).where(
            UserSession.user_id == str(current_user.id),
            UserSession.session_remark == "临时会话"
        ).order_by(UserSession.upload_time.desc()).limit(1)
        
        result = await session.execute(query)
        existing_session = result.scalar_one_or_none()
        
        if existing_session:
            # 找到现有临时会话，直接返回
            return {
                "success": True,
                "session_id": existing_session.id,
                "session_title": existing_session.session_title,
                "dataset_id": existing_session.dataset_ids,
                "is_new": False,
                "message": "找到现有临时会话"
            }
        else:
            # 没找到临时会话，创建新的临时会话
            session_title = "临时会话"
            new_dataset_ids = ""
            
            # 如果提供了数据集ID，验证并设置
            if dataset_id:
                dataset_query = select(DataSet).where(DataSet.id == dataset_id)
                dataset_result = await session.execute(dataset_query)
                dataset = dataset_result.scalar_one_or_none()
                
                if not dataset:
                    raise HTTPException(
                        status_code=404,
                        detail="数据集不存在"
                    )
                
                new_dataset_ids = dataset_id
                session_title = f"临时会话 - {dataset.name}"
            
            # 创建新会话
            new_session = UserSession(
                user_id=str(current_user.id),
                file_name=session_title,
                session_title=session_title,
                file_url="临时会话",
                dataset_ids=new_dataset_ids,
                session_remark="临时会话",  # 标记为临时会话
                upload_time=datetime.now()
            )
            
            # 保存会话
            session.add(new_session)
            await session.commit()
            await session.refresh(new_session)
            
            return {
                "success": True,
                "session_id": new_session.id,
                "session_title": new_session.session_title,
                "dataset_id": dataset_id,
                "is_new": True,
                "message": "创建新的临时会话"
            }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取或创建临时会话失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(
            status_code=500,
            detail=f"获取或创建临时会话失败: {str(e)}"
        )

@router.post("/get_session_messages", response_model=dict)
async def get_session_messages(
    session_id: str = Form(..., description="会话ID"),
    page: int = Form(1, ge=1, description="页码，从1开始"),
    page_size: int = Form(20, ge=1, le=100, description="每页消息数量，最大100"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    根据会话ID分页获取历史消息
    
    Args:
        session_id: 会话ID
        page: 页码，从1开始
        page_size: 每页消息数量，默认20，最大100
        
    Returns:
        dict: 包含消息列表和分页信息的响应
    """
    try:
        # 1. 验证会话是否存在且属于当前用户
        session_query = select(UserSession).where(
            UserSession.id == session_id,
            UserSession.user_id == str(current_user.id)
        )
        session_result = await session.execute(session_query)
        user_session = session_result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(
                status_code=404,
                detail="会话不存在或无权限访问"
            )
        
        # 2. 构建消息查询
        messages_query = select(UserMsg).where(
            UserMsg.session_id == session_id
        ).order_by(UserMsg.timestamp.desc())  # 按时间倒序，最新的在前
        
        # 3. 计算总数
        count_query = select(func.count()).select_from(
            select(UserMsg).where(UserMsg.session_id == session_id).subquery()
        )
        total_count = await session.scalar(count_query)
        
        # 4. 计算分页参数
        offset = (page - 1) * page_size
        total_pages = (total_count + page_size - 1) // page_size  # 向上取整
        
        # 5. 执行分页查询
        paginated_query = messages_query.offset(offset).limit(page_size)
        messages_result = await session.execute(paginated_query)
        messages = messages_result.scalars().all()

        # 7. 构建响应
        return {
            "success": True,
            "data": messages,
            "pagination": {
                "page": page,
                "page_size": page_size,
                "total_count": total_count,
                "total_pages": total_pages,
                "has_previous": page > 1,
                "has_next": page < total_pages,
                "previous_page": page - 1 if page > 1 else None,
                "next_page": page + 1 if page < total_pages else None
            },
            "session_info": {
                "session_id": session_id,
                "session_title": user_session.session_title,
                "upload_time": user_session.upload_time.isoformat() if user_session.upload_time else None
            }
        }
    
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取会话消息失败: {str(e)}")
        logger.exception(e)
        raise HTTPException(
            status_code=500,
            detail=f"获取会话消息失败: {str(e)}"
        )

@router.post("/update_session_datasets", response_model=dict)
async def update_session_datasets(
    session_id: str = Form(..., description="会话ID"),
    dataset_ids: str = Form(..., description="数据集ID列表，多个ID用逗号分隔，传空字符串表示清空关联"),
    current_user: User = Depends(current_active_user),
    session: AsyncSession = Depends(get_async_session)
):
    """
    更新会话关联的数据集
    
    Args:
        session_id: 会话ID
        dataset_ids: 数据集ID列表，多个ID用逗号分隔，传空字符串表示清空关联
    """
    try:
        # 1. 查找会话并验证权限
        query = select(UserSession).where(
            UserSession.id == session_id,
            UserSession.user_id == str(current_user.id)
        )
        result = await session.execute(query)
        user_session = result.scalar_one_or_none()
        
        if not user_session:
            raise HTTPException(
                status_code=404,
                detail="会话不存在或无权限访问"
            )
        
        # 2. 处理数据集ID列表
        user_session.dataset_ids = dataset_ids
        # 3. 更新会话
        user_session.upload_time = datetime.now()  # 更新时间戳
        await session.commit()
    
        return {
            "status": True,
            "message": "会话数据集关联更新成功"
        }
    
    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"更新会话数据集关联失败: {str(e)}"
        )

@router.post("/get_message_task_data", response_model=dict)
async def get_message_task_data(
    message_id: str = Form(..., description="消息ID"),
    current_user: User = Depends(current_active_user),
    db: AsyncSession = Depends(get_async_session)
):
    """
    根据消息ID获取关联任务的数据
    
    Args:
        message_id: 消息ID
        current_user: 当前用户
        db: 数据库会话
        
    Returns:
        dict: 包含任务数据的响应
    """
    try:
        # 首先查询消息是否存在且属于当前用户
        msg_query = select(UserMsg).where(
            UserMsg.id == message_id,
            UserMsg.user_id == str(current_user.id)
        )
        msg_result = await db.execute(msg_query)
        user_msg = msg_result.scalar_one_or_none()
        
        if not user_msg:
            raise HTTPException(status_code=404, detail="消息不存在或无权限访问")
        
        # 获取关联的任务数据 - 只从dk_tasks表中查询
        tasks_data = []
        
        # 通过Task表的messageid字段查询关联任务
        tasks_query = select(Task).where(Task.messageid == message_id)
        tasks_result = await db.execute(tasks_query)
        tasks = tasks_result.scalars().all()
        
        for task in tasks:
            # 解析result_data
            result_data = task.get_result_data()
            
            tasks_data.append({
                "task_id": task.id,
                "task_name": task.task_name,
                "task_type": task.task_type,
                "status": task.status,
                "execution_status": task.execution_status,
                "result_data": result_data,
                "columns_info": task.get_columns_info(),
                "sql_query": task.sql_query,
                "rows_affected": task.rows_affected,
                "duration_seconds": task.get_duration_seconds(),
                "created_at": task.created_at.isoformat() if task.created_at else None,
                "completed_at": task.completed_at.isoformat() if task.completed_at else None,
                "error_message": task.error_message
            })
        
        # 返回结果
        return {
            "success": True,
            "message": "获取任务数据成功",
            "data": {
                "message_info": {
                    "id": user_msg.id,
                    "message_type": user_msg.message_type,
                    "message": user_msg.message,
                    "timestamp": user_msg.timestamp.isoformat(),
                    "message_sourceid": user_msg.message_sourceid
                },
                "tasks": tasks_data,
                "total_tasks": len(tasks_data)
            }
        }
        
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"获取消息任务数据失败: {str(e)}")
        raise HTTPException(status_code=500, detail=f"获取任务数据失败: {str(e)}")
